Source code for activity.models.activity

from __future__ import annotations

from onegov.activity.models.occasion import Occasion
from onegov.activity.models.period import Period
from onegov.activity.utils import extract_thumbnail, extract_municipality
from onegov.core.orm import Base, observes
from onegov.core.orm.mixins import (
    dict_markup_property,
    ContentMixin,
    meta_property,
    TimestampMixin,
)
from onegov.core.orm.types import UUID
from onegov.core.utils import normalize_for_url
from onegov.user import User
from sqlalchemy import Column, Enum, Text, ForeignKey
from sqlalchemy import exists, and_, desc
from sqlalchemy.dialects.postgresql import HSTORE
from sqlalchemy.ext.mutable import MutableDict
from sqlalchemy.orm import object_session, relationship
from uuid import uuid4


from typing import Any, TYPE_CHECKING
if TYPE_CHECKING:
    import uuid
    from collections.abc import Iterable
    from onegov.activity.collections import PublicationRequestCollection
    from onegov.activity.models import PeriodMeta, PublicationRequest
    from onegov.core.orm.mixins import dict_property
    from typing import Literal
    from typing import Self, TypeAlias

[docs] ActivityState: TypeAlias = Literal[ 'preview', 'proposed', 'accepted', 'archived' ]
# Note, a database migration is needed if these states are changed
[docs] ACTIVITY_STATES: tuple[ActivityState, ...] = ( 'preview', 'proposed', 'accepted', 'archived' )
[docs] class Activity(Base, ContentMixin, TimestampMixin): """ Describes an activity that is made available to participants on certain occasions (i.e. dates). The activity describes the what's going on, the occasion describes when and with whom. """
[docs] __tablename__ = 'activities'
#: An internal id for references (not public)
[docs] id: Column[uuid.UUID] = Column( UUID, # type:ignore[arg-type] primary_key=True, default=uuid4 )
#: A nice id for the url, readable by humans
[docs] name: Column[str] = Column(Text, nullable=False, unique=True)
#: The title of the activity
[docs] title: Column[str] = Column(Text, nullable=False)
#: The normalized title for sorting
[docs] order: Column[str] = Column(Text, nullable=False, index=True)
#: Describes the activity briefly
[docs] lead: dict_property[str | None] = meta_property()
#: Describes the activity in detail
[docs] text = dict_markup_property('content')
#: The thumbnail shown in the overview
[docs] thumbnail: dict_property[str | None] = meta_property()
#: Tags/Categories of the activity
[docs] _tags: Column[dict[str, str] | None] = Column( # type:ignore MutableDict.as_mutable(HSTORE), # type:ignore[no-untyped-call] nullable=True, name='tags' )
#: The user to which this activity belongs to (organiser)
[docs] username: Column[str] = Column( Text, ForeignKey(User.username), nullable=False )
#: The user which initially reported this activity (same as username, but #: this value may not change after initialisation)
[docs] reporter: Column[str] = Column(Text, nullable=False)
#: Describes the location of the activity
[docs] location: Column[str | None] = Column(Text, nullable=True)
#: The municipality in which the activity is held, from the location
[docs] municipality: Column[str | None] = Column(Text, nullable=True)
#: Access the user linked to this activity
[docs] user: relationship[User] = relationship(User)
#: The occasions linked to this activity
[docs] occasions: relationship[list[Occasion]] = relationship( Occasion, order_by='Occasion.order', back_populates='activity' )
#: the type of the item, this can be used to create custom polymorphic #: subclasses of this class. See #: `<http://docs.sqlalchemy.org/en/improve_toc/\ #: orm/extensions/declarative/inheritance.html>`_.
[docs] type: Column[str] = Column( Text, nullable=False, default=lambda: 'generic' )
#: the state of the activity
[docs] state: Column[ActivityState] = Column( Enum(*ACTIVITY_STATES, name='activity_state'), # type:ignore[arg-type] nullable=False, default='preview' )
#: The publication requests linked to this activity
[docs] publication_requests: relationship[list[PublicationRequest]]
publication_requests = relationship( 'PublicationRequest', back_populates='activity' )
[docs] __mapper_args__ = { 'polymorphic_on': 'type', 'polymorphic_identity': 'generic', }
@observes('title')
[docs] def title_observer(self, title: str) -> None: self.order = normalize_for_url(title)
@observes('username')
[docs] def username_observer(self, username: str) -> None: if not self.reporter: self.reporter = username
@observes('content')
[docs] def content_observer(self, content: dict[str, Any] | None) -> None: self.thumbnail = extract_thumbnail(self.content.get('text'))
@observes('location')
[docs] def location_observer(self, content: str | None) -> None: municipality = extract_municipality(self.location) if municipality: self.municipality = municipality[1] else: self.municipality = None
@property
[docs] def tags(self) -> set[str]: return set(self._tags.keys()) if self._tags else set()
# FIXME: asymmetric property @tags.setter def tags(self, value: Iterable[str]) -> None: self._tags = dict.fromkeys(value, '') if value else None
[docs] def propose(self) -> Self: assert self.state in ('preview', 'proposed') self.state = 'proposed' return self
[docs] def accept(self) -> Self: self.state = 'accepted' return self
[docs] def archive(self) -> Self: self.state = 'archived' return self
[docs] def create_publication_request( self, period: Period, **kwargs: Any # TODO: better type safety ) -> PublicationRequest: return self.requests.add(activity=self, period=period, **kwargs)
@property
[docs] def requests(self) -> PublicationRequestCollection: # XXX circular imports from onegov.activity.collections.publication_request import ( PublicationRequestCollection) return PublicationRequestCollection(object_session(self))
@property
[docs] def latest_request(self) -> PublicationRequest | None: q = self.requests.query() q = q.filter_by(activity_id=self.id) q = q.join(Period) q = q.order_by(desc(Period.active), desc(Period.execution_start)) return q.first()
[docs] def request_by_period( self, period: Period | PeriodMeta | None ) -> PublicationRequest | None: if not period: return None q = self.requests.query() q = q.filter_by(activity_id=self.id, period_id=period.id) return q.first()
[docs] def has_occasion_in_period(self, period: Period | PeriodMeta) -> bool: q = object_session(self).query( exists().where(and_( Occasion.activity_id == self.id, Occasion.period_id == period.id )) ) return q.scalar()