from __future__ import annotations
from functools import cached_property
from onegov.people import AgencyMembershipCollection
from typing import Protocol
from typing import TYPE_CHECKING
if TYPE_CHECKING:
from onegov.agency.collections import ExtendedAgencyCollection
from onegov.agency.models import ExtendedAgency # noqa: F401
from onegov.core.orm import Base
from onegov.core.orm.abstract import MoveDirection
from onegov.people import AgencyMembership # noqa: F401
from sqlalchemy.orm import Session
from uuid import UUID
[docs]
class SupportsById[M_co, IdT_contra](Protocol):
[docs]
def by_id(self, id: IdT_contra, /) -> M_co | None: ...
[docs]
class Move[M: Base, IdT_contra: UUID | int]:
""" Base class for moving things. """
def __init__(
self,
session: Session,
subject_id: IdT_contra,
target_id: IdT_contra,
direction: MoveDirection
) -> None:
[docs]
self.subject_id = subject_id
[docs]
self.target_id = target_id
[docs]
self.direction = direction
@cached_property
[docs]
def collection(self) -> SupportsById[M, IdT_contra]:
raise NotImplementedError
@cached_property
[docs]
def subject(self) -> M | None:
return self.collection.by_id(self.subject_id)
@cached_property
[docs]
def target(self) -> M | None:
return self.collection.by_id(self.target_id)
[docs]
def execute(self) -> None:
raise NotImplementedError
[docs]
class AgencyMove(Move['ExtendedAgency', int]):
""" Represents a single move of a suborganization. """
@cached_property
[docs]
def collection(self) -> ExtendedAgencyCollection:
from onegov.agency.collections import ExtendedAgencyCollection
return ExtendedAgencyCollection(self.session)
[docs]
def execute(self) -> None:
if self.subject and self.target and self.subject != self.target:
if self.subject.parent_id == self.target.parent_id:
self.collection.move(
subject=self.subject,
target=self.target,
direction=self.direction
)
[docs]
class AgencyMembershipMoveWithinAgency(Move['AgencyMembership', 'UUID']):
""" Represents a single move of a membership with respect to a Agency. """
@cached_property
[docs]
def collection(self) -> AgencyMembershipCollection:
return AgencyMembershipCollection(self.session)
[docs]
def execute(self) -> None:
if self.subject and self.target and self.subject != self.target:
if self.subject.agency_id == self.target.agency_id:
self.collection.move(
subject=self.subject,
target=self.target,
direction=self.direction,
move_on_col='order_within_agency'
)
[docs]
class AgencyMembershipMoveWithinPerson(Move['AgencyMembership', 'UUID']):
""" Represents a single move of a membership with respect to a Person. """
@cached_property
[docs]
def collection(self) -> AgencyMembershipCollection:
return AgencyMembershipCollection(self.session)
[docs]
def execute(self) -> None:
if self.subject and self.target and self.subject != self.target:
if self.subject.person_id == self.target.person_id:
self.collection.move(
subject=self.subject,
target=self.target,
direction=self.direction,
move_on_col='order_within_person'
)