Skip to content
New issue

Have a question about this project? Sign up for a free GitHub account to open an issue and contact its maintainers and the community.

By clicking “Sign up for GitHub”, you agree to our terms of service and privacy statement. We’ll occasionally send you account related emails.

Already on GitHub? Sign in to your account

[#87] Added ScheduleConstraint enum. #126

Merged
merged 3 commits into from
Dec 10, 2024
Merged
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
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