from onegov.core.collection import GenericCollection
from onegov.core.utils import normalize_for_url, increment_name, is_uuid
from onegov.directory.models import Directory
from onegov.directory.models.directory import EntryRecipient
from onegov.directory.types import DirectoryConfiguration
from typing import overload, Any, Literal, TypeVar, TYPE_CHECKING
if TYPE_CHECKING:
from sqlalchemy.orm import Query, Session
from uuid import UUID
[docs]
DirectoryT = TypeVar('DirectoryT', bound=Directory)
[docs]
class DirectoryCollection(GenericCollection[DirectoryT]):
@overload
def __init__(
self: 'DirectoryCollection[Directory]',
session: 'Session',
type: Literal['*', 'generic'] = '*'
) -> None: ...
@overload
def __init__(self, session: 'Session', type: str) -> None: ...
def __init__(self, session: 'Session', type: str = '*') -> None:
super().__init__(session)
@property
[docs]
def model_class(self) -> type[DirectoryT]:
return Directory.get_polymorphic_class( # type:ignore[return-value]
self.type,
Directory # type:ignore[arg-type]
)
[docs]
def query(self) -> 'Query[DirectoryT]':
return super().query().order_by(self.model_class.order)
[docs]
def add(self, **kwargs: Any) -> DirectoryT:
if self.type != '*':
kwargs.setdefault('type', self.type)
kwargs['name'] = self.unique_name(kwargs['title'])
if 'configuration' not in kwargs:
kwargs['configuration'] = DirectoryConfiguration()
elif isinstance(kwargs['configuration'], str):
kwargs['configuration'] = DirectoryConfiguration.from_yaml(
kwargs['configuration']
)
return super().add(**kwargs)
[docs]
def unique_name(self, title: str) -> str:
names = {n for n, in self.session.query(self.model_class.name)}
name = normalize_for_url(title)
# add an upper limit to how many times increment_name can fail
# to find a suitable name
for _ in range(100):
if name not in names:
return name
name = increment_name(name)
raise RuntimeError('Increment name failed to find a candidate')
[docs]
def by_name(self, name: str) -> DirectoryT | None:
return self.query().filter_by(name=name).first()
[docs]
class EntryRecipientCollection:
def __init__(self, session: 'Session'):
[docs]
self.session = session
[docs]
def query(self) -> 'Query[EntryRecipient]':
return self.session.query(EntryRecipient)
[docs]
def by_id(self, id: 'str | UUID') -> EntryRecipient | None:
if is_uuid(id):
return self.query().filter(EntryRecipient.id == id).first()
return None
[docs]
def add(
self,
address: str,
directory_id: 'UUID',
confirmed: bool = False
) -> EntryRecipient:
recipient = EntryRecipient(
address=address,
directory_id=directory_id,
confirmed=confirmed
)
self.session.add(recipient)
self.session.flush()
return recipient
[docs]
def delete(self, recipient: EntryRecipient) -> None:
self.session.delete(recipient)
self.session.flush()