Source code for dokomoforms.models.user

"""User models."""

from collections import OrderedDict

from sqlalchemy.dialects import postgresql as pg
from sqlalchemy.orm import relationship
from sqlalchemy.sql.functions import current_timestamp
import sqlalchemy as sa

from dokomoforms.models import util, Base


[docs]class User(Base): """Models a user. A user has at least one e-mail address.""" __tablename__ = 'auth_user' id = util.pk() name = sa.Column(pg.TEXT, nullable=False) emails = relationship( 'Email', order_by='Email.address', backref='user', cascade='all, delete-orphan', passive_deletes=True, ) role = sa.Column( sa.Enum( 'enumerator', 'administrator', name='user_roles', inherit_schema=True ), nullable=False, ) preferences = util.json_column( 'preferences', default='{"default_language": "English"}' ) last_update_time = util.last_update_time() __mapper_args__ = { 'polymorphic_identity': 'enumerator', 'polymorphic_on': role, } __table_args__ = ( sa.CheckConstraint( "((preferences->>'default_language')) IS NOT NULL", name='must_specify_default_language' ), ) def _asdict(self) -> OrderedDict: return OrderedDict(( ('id', self.id), ('deleted', self.deleted), ('name', self.name), ('emails', [email.address for email in self.emails]), ('role', self.role), ('preferences', self.preferences), ('allowed_surveys', [s.id for s in self.allowed_surveys]), ('last_update_time', self.last_update_time), ))
[docs]class Administrator(User): """A User who can create Surveys and add Users. Regular users can answer surveys, but only Administrator instances can create surveys. """ __tablename__ = 'administrator' id = util.pk('auth_user.id') surveys = relationship( 'Survey', order_by='Survey.created_on', backref='creator', cascade='all, delete-orphan', passive_deletes=True, ) token = sa.Column(pg.BYTEA) token_expiration = sa.Column( pg.TIMESTAMP(timezone=True), nullable=False, server_default=current_timestamp(), ) __mapper_args__ = {'polymorphic_identity': 'administrator'} def _asdict(self) -> OrderedDict: result = super()._asdict() result['surveys'] = [s.id for s in self.surveys] result['admin_surveys'] = [s.id for s in self.admin_surveys] result['token_expiration'] = self.token_expiration return result
[docs]def construct_user(*, role: str, **kwargs): """Construct either an enumerator or an administrator.""" if role == 'enumerator': user_constructor = User elif role == 'administrator': user_constructor = Administrator else: raise TypeError return user_constructor(**kwargs)
[docs]class Email(Base): """Models an e-mail address.""" __tablename__ = 'email' id = util.pk() address = sa.Column( pg.TEXT, sa.CheckConstraint("address ~ '.*@.*'"), nullable=False, unique=True ) user_id = sa.Column(pg.UUID, util.fk('auth_user.id'), nullable=False) last_update_time = util.last_update_time() def _asdict(self) -> OrderedDict: return OrderedDict(( ('id', self.id), ('address', self.address), ('user', self.user.name), ('last_update_time', self.last_update_time), ))