Skip to content

Commit

Permalink
Merge pull request #126 from eoyilmaz/87-create-scheduleconstraint-enum
Browse files Browse the repository at this point in the history
[#87] Added `ScheduleConstraint` enum.
  • Loading branch information
eoyilmaz authored Dec 10, 2024
2 parents 155b6d1 + 95baf79 commit a95a734
Show file tree
Hide file tree
Showing 8 changed files with 359 additions and 144 deletions.
1 change: 0 additions & 1 deletion src/stalker/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -264,7 +264,6 @@ class Config(ConfigBase):
review_status_codes=["NEW", "RREV", "APP"],
daily_status_names=["Open", "Closed"],
daily_status_codes=["OPEN", "CLS"],
task_schedule_constraints=["none", "start", "end", "both"],
task_schedule_models=["effort", "length", "duration"],
task_dependency_gap_models=["length", "duration"],
task_dependency_targets=["onend", "onstart"],
Expand Down
5 changes: 3 additions & 2 deletions src/stalker/db/types.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,6 @@
# -*- coding: utf-8 -*-
"""Stalker specific data types are situated here."""
import datetime
import json
from typing import Any, Dict, TYPE_CHECKING, Union

Expand Down Expand Up @@ -65,7 +66,7 @@ class DateTimeUTC(TypeDecorator):

impl = DateTime

def process_bind_param(self, value, dialect):
def process_bind_param(self, value: Any, dialect: str) -> datetime.datetime:
"""Process bind param.
Args:
Expand All @@ -81,7 +82,7 @@ def process_bind_param(self, value, dialect):
value = value.astimezone(pytz.utc)
return value

def process_result_value(self, value, dialect):
def process_result_value(self, value: Any, dialect: str) -> datetime.datetime:
"""Process result value.
Args:
Expand Down
142 changes: 106 additions & 36 deletions src/stalker/models/mixins.py
Original file line number Diff line number Diff line change
Expand Up @@ -2,6 +2,7 @@
"""Mixins are situated here."""

import datetime
from enum import IntEnum
from typing import (
Any,
Dict,
Expand All @@ -17,7 +18,7 @@

import pytz

from sqlalchemy import Column, Enum, Float, ForeignKey, Integer, Interval, String, Table
from sqlalchemy import Column, Enum, Float, ForeignKey, Integer, Interval, String, Table, TypeDecorator
from sqlalchemy.exc import OperationalError, UnboundExecutionError
from sqlalchemy.ext.declarative import declared_attr
from sqlalchemy.orm import (
Expand Down Expand Up @@ -173,6 +174,94 @@ def create_secondary_table(
return secondary_table


class ScheduleConstraint(IntEnum):
"""The schedule constraint enum."""
NONE = 0
Start = 1
End = 2
Both = 3

def __repr__(self) -> str:
"""Return the enum name for str().
Returns:
str: The name as the string representation of this
ScheduleConstraint.
"""
return self.name if self.name != "NONE" else "None"

__str__ = __repr__

@classmethod
def to_constraint(cls, constraint: Union[int, str, "ScheduleConstraint"]) -> Self:
"""Validate and return type enum from an input int or str value.
Args:
constraint (Union[str, ScheduleConstraint]): Input `constraint` value.
quiet (bool): To raise any exception for invalid value.
Raises:
TypeError: Input value type is invalid.
ValueError: Input value is invalid.
Returns:
ScheduleConstraint: ScheduleConstraint value.
"""
# Check if it's a valid str type for a constraint.
if constraint is None:
constraint = ScheduleConstraint.NONE

if not isinstance(constraint, (int, str, ScheduleConstraint)):
raise TypeError(
"constraint should be an int, str or ScheduleConstraint, "
f"not {constraint.__class__.__name__}: '{constraint}'"
)

if isinstance(constraint, str):
constraint_name_lut = dict([(e.name.lower(), e.name.title() if e.name != "NONE" else "NONE") for e in cls])
# also add int values
constraint_lower_case = constraint.lower()
if constraint_lower_case not in constraint_name_lut:
raise ValueError(
"constraint should be one of {}, not '{}'".format(
[e.name.title() for e in cls], constraint
)
)

# Return the enum status for the status value.
return cls.__members__[constraint_name_lut[constraint_lower_case]]
else:
return ScheduleConstraint(constraint)


class ScheduleConstraintDecorator(TypeDecorator):
"""Store ScheduleConstraint as an integer and restore as ScheduleConstraint."""

impl = Integer

def process_bind_param(self, value, dialect) -> int:
"""Return the integer value of the ScheduleConstraint.
Args:
value (ScheduleConstraint): The ScheduleConstraint value.
dialect (str): The name of the dialect.
Returns:
int: The value of the ScheduleConstraint.
"""
# just return the value
return value.value

def process_result_value(self, value, dialect):
"""Return a ScheduleConstraint.
Args:
value (int): The integer value.
dialect (str): The name of the dialect.
"""
return ScheduleConstraint.to_constraint(value)


class TargetEntityTypeMixin(object):
"""Adds target_entity_type attribute to mixed in class.
Expand Down Expand Up @@ -1338,7 +1427,7 @@ def __init__(
schedule_timing: Optional[float] = None,
schedule_unit: Optional[str] = None,
schedule_model: Optional[str] = None,
schedule_constraint: int = 0,
schedule_constraint: ScheduleConstraint = ScheduleConstraint.NONE,
**kwargs: Dict[str, Any],
) -> None:
self.schedule_constraint = schedule_constraint
Expand Down Expand Up @@ -1440,19 +1529,19 @@ def schedule_model(cls) -> Mapped[str]:
)

@declared_attr
def schedule_constraint(cls) -> Mapped[int]:
def schedule_constraint(cls) -> Mapped[ScheduleConstraint]:
"""Create the schedule_constraint attribute as a declared attribute.
Returns:
Column: The Column related to the schedule_constraint attribute.
"""
return mapped_column(
f"{cls.__default_schedule_attr_name__}_constraint",
Integer,
ScheduleConstraintDecorator(),
default=0,
nullable=False,
doc="""An integer number showing the constraint schema for this
task.
doc="""A ScheduleConstraint value showing the constraint schema
for this task.
Possible values are:
Expand All @@ -1463,59 +1552,40 @@ def schedule_constraint(cls) -> Mapped[int]:
3 Constrain Both
===== ===============
For convenience use **stalker.models.task.CONSTRAIN_NONE**,
**stalker.models.task.CONSTRAIN_START**,
**stalker.models.task.CONSTRAIN_END**,
**stalker.models.task.CONSTRAIN_BOTH**.
This value is going to be used to constrain the start and end date
values of this task. So if you want to pin the start of a task to a
certain date. Set its :attr:`.schedule_constraint` value to
**CONSTRAIN_START**. When the task is scheduled by **TaskJuggler**
the start date will be pinned to the :attr:`start` attribute of
this task.
:attr:`.ScheduleConstraint.Start`. When the task is scheduled by
**TaskJuggler** the start date will be pinned to the :attr:`start`
attribute of this task.
And if both of the date values (start and end) wanted to be pinned
to certain dates (making the task effectively a ``duration`` task)
set the desired :attr:`start` and :attr:`end` and then set the
:attr:`schedule_constraint` to **CONSTRAIN_BOTH**.
:attr:`schedule_constraint` to :att:`.ScheduleConstraint.Both`.
""",
)

@validates("schedule_constraint")
def _validate_schedule_constraint(
self,
key: str,
schedule_constraint: Union[None, int],
schedule_constraint: Union[None, int, str],
) -> int:
"""Validate the given schedule_constraint value.
Args:
key (str): The name of the validated column.
schedule_constraint (int): The schedule_constraint value to be validated.
Raises:
TypeError: If the schedule_constraint is not an int.
schedule_constraint (Union[None, int, str]): The value to be
validated.
Returns:
int: The validated schedule_constraint value.
ScheduleConstraint: The validated schedule_constraint value.
"""
if not schedule_constraint:
schedule_constraint = 0

if not isinstance(schedule_constraint, int):
raise TypeError(
"{cls}.{attr}_constraint should be an integer "
"between 0 and 3, not {constraint_class}: '{constraint}'".format(
cls=self.__class__.__name__,
attr=self.__default_schedule_attr_name__,
constraint_class=schedule_constraint.__class__.__name__,
constraint=schedule_constraint,
)
)
if schedule_constraint is None:
schedule_constraint = ScheduleConstraint.NONE

schedule_constraint = max(schedule_constraint, 0)
schedule_constraint = min(schedule_constraint, 3)
schedule_constraint = ScheduleConstraint.to_constraint(schedule_constraint)

return schedule_constraint

Expand Down
Loading

0 comments on commit a95a734

Please sign in to comment.