diff --git a/funnel/models/account.py b/funnel/models/account.py index 974e0ecb3..6ef2a3728 100644 --- a/funnel/models/account.py +++ b/funnel/models/account.py @@ -259,7 +259,7 @@ class Account(UuidMixin, BaseMixin, Model): sa.orm.mapped_column(sa.Integer, nullable=False), read={'all'} ) - search_vector: Mapped[TSVectorType] = sa.orm.mapped_column( + search_vector: Mapped[str] = sa.orm.mapped_column( TSVectorType( 'title', 'name', @@ -1424,7 +1424,7 @@ class Team(UuidMixin, BaseMixin, Model): account_id: Mapped[int] = sa.orm.mapped_column( sa.ForeignKey('account.id'), nullable=False, index=True ) - account = with_roles( + account: Mapped[Account] = with_roles( relationship( Account, foreign_keys=[account_id], diff --git a/funnel/models/account_membership.py b/funnel/models/account_membership.py index 80c4b97b7..23faa8be8 100644 --- a/funnel/models/account_membership.py +++ b/funnel/models/account_membership.py @@ -151,10 +151,12 @@ class __Account: ) owner_users = with_roles( - DynamicAssociationProxy('active_owner_memberships', 'member'), read={'all'} + DynamicAssociationProxy[Account]('active_owner_memberships', 'member'), + read={'all'}, ) admin_users = with_roles( - DynamicAssociationProxy('active_admin_memberships', 'member'), read={'all'} + DynamicAssociationProxy[Account]('active_admin_memberships', 'member'), + read={'all'}, ) # pylint: disable=invalid-unary-operand-type diff --git a/funnel/models/auth_client.py b/funnel/models/auth_client.py index cf8281010..8c1dc8b40 100644 --- a/funnel/models/auth_client.py +++ b/funnel/models/auth_client.py @@ -105,24 +105,24 @@ class AuthClient(ScopeMixin, UuidMixin, BaseMixin, Model): grants_via={None: {'owner': 'owner', 'admin': 'admin'}}, ) #: Human-readable title - title = with_roles( + title: Mapped[str] = with_roles( sa.orm.mapped_column(sa.Unicode(250), nullable=False), read={'all'}, write={'owner'}, ) #: Long description - description = with_roles( + description: Mapped[str] = with_roles( sa.orm.mapped_column(sa.UnicodeText, nullable=False, default=''), read={'all'}, write={'owner'}, ) #: Confidential or public client? Public has no secret key - confidential = with_roles( + confidential: Mapped[bool] = with_roles( sa.orm.mapped_column(sa.Boolean, nullable=False), read={'all'}, write={'owner'} ) #: Website - website = with_roles( - sa.orm.mapped_column(sa.UnicodeText, nullable=False), + website: Mapped[str] = with_roles( + sa.orm.mapped_column(sa.UnicodeText, nullable=False), # FIXME: Use UrlType read={'all'}, write={'owner'}, ) @@ -131,7 +131,7 @@ class AuthClient(ScopeMixin, UuidMixin, BaseMixin, Model): 'redirect_uri', sa.UnicodeText, nullable=True, default='' ) #: Back-end notification URI (TODO: deprecated, needs better architecture) - notification_uri = with_roles( + notification_uri: Mapped[str | None] = with_roles( # FIXME: Use UrlType sa.orm.mapped_column(sa.UnicodeText, nullable=True, default=''), rw={'owner'} ) #: Active flag @@ -139,7 +139,7 @@ class AuthClient(ScopeMixin, UuidMixin, BaseMixin, Model): sa.Boolean, nullable=False, default=True ) #: Allow anyone to login to this app? - allow_any_login = with_roles( + allow_any_login: Mapped[bool] = with_roles( sa.orm.mapped_column(sa.Boolean, nullable=False, default=True), read={'all'}, write={'owner'}, @@ -150,7 +150,7 @@ class AuthClient(ScopeMixin, UuidMixin, BaseMixin, Model): #: as a trusted client to provide single sign-in across the services. #: However, resources in the scope column (via ScopeMixin) are granted for #: any arbitrary user without explicit user authorization. - trusted = with_roles( + trusted: Mapped[bool] = with_roles( sa.orm.mapped_column(sa.Boolean, nullable=False, default=False), read={'all'} ) @@ -426,7 +426,7 @@ class AuthToken(ScopeMixin, BaseMixin, Model): backref=backref('authtokens', lazy='dynamic', cascade='all'), ) #: The session in which this token was issued, null for confidential clients - login_session_id = sa.orm.mapped_column( + login_session_id: Mapped[int | None] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('login_session.id'), nullable=True ) login_session: Mapped[LoginSession | None] = with_roles( @@ -434,7 +434,7 @@ class AuthToken(ScopeMixin, BaseMixin, Model): read={'owner'}, ) #: The client this authtoken is for - auth_client_id = sa.orm.mapped_column( + auth_client_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('auth_client.id'), nullable=False, index=True ) auth_client: Mapped[AuthClient] = with_roles( @@ -445,19 +445,23 @@ class AuthToken(ScopeMixin, BaseMixin, Model): read={'owner'}, ) #: The token - token = sa.orm.mapped_column( + token: Mapped[str] = sa.orm.mapped_column( sa.String(22), default=make_buid, nullable=False, unique=True ) #: The token's type, 'bearer', 'mac' or a URL - token_type = sa.orm.mapped_column(sa.String(250), default='bearer', nullable=False) + token_type: Mapped[str] = sa.orm.mapped_column( + sa.String(250), default='bearer', nullable=False + ) #: Token secret for 'mac' type - secret = sa.orm.mapped_column(sa.String(44), nullable=True) + secret: Mapped[str | None] = sa.orm.mapped_column(sa.String(44), nullable=True) #: Secret's algorithm (for 'mac' type) - algorithm = sa.orm.mapped_column(sa.String(20), nullable=True) + algorithm: Mapped[str | None] = sa.orm.mapped_column(sa.String(20), nullable=True) #: Token's validity period in seconds, 0 = unlimited - validity = sa.orm.mapped_column(sa.Integer, nullable=False, default=0) + validity: Mapped[int] = sa.orm.mapped_column(sa.Integer, nullable=False, default=0) #: Refresh token, to obtain a new token - refresh_token = sa.orm.mapped_column(sa.String(22), nullable=True, unique=True) + refresh_token: Mapped[str | None] = sa.orm.mapped_column( + sa.String(22), nullable=True, unique=True + ) # Only one authtoken per user and client. Add to scope as needed __table_args__ = ( @@ -472,13 +476,6 @@ class AuthToken(ScopeMixin, BaseMixin, Model): } } - @property - def effective_user(self) -> Account: - """Return subject user of this auth token.""" - if self.login_session: - return self.login_session.account - return self.account - def __init__(self, **kwargs) -> None: super().__init__(**kwargs) self.token = make_buid() @@ -491,13 +488,20 @@ def __repr__(self) -> str: return f'' @property - def effective_scope(self) -> list: + def effective_user(self) -> Account: + """Return subject user of this auth token.""" + if self.login_session: + return self.login_session.account + return cast(Account, self.account) + + @property + def effective_scope(self) -> list[str]: """Return effective scope of this token, combining granted and client scopes.""" return sorted(set(self.scope) | set(self.auth_client.scope)) @with_roles(read={'owner'}) @cached_property - def last_used(self) -> datetime: + def last_used(self) -> datetime | None: """Return last used timestamp for this auth token.""" return ( db.session.query(sa.func.max(auth_client_login_session.c.accessed_at)) @@ -645,7 +649,7 @@ class AuthClientPermissions(BaseMixin, Model): backref=backref('client_permissions', cascade='all'), ) #: AuthClient app they are assigned on - auth_client_id = sa.orm.mapped_column( + auth_client_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('auth_client.id'), nullable=False, index=True ) auth_client: Mapped[AuthClient] = with_roles( @@ -657,7 +661,7 @@ class AuthClientPermissions(BaseMixin, Model): grants_via={None: {'owner'}}, ) #: The permissions as a string of tokens - access_permissions = sa.orm.mapped_column( + access_permissions: Mapped[str] = sa.orm.mapped_column( 'permissions', sa.UnicodeText, default='', nullable=False ) @@ -715,14 +719,16 @@ class AuthClientTeamPermissions(BaseMixin, Model): __tablename__ = 'auth_client_team_permissions' #: Team which has these permissions - team_id = sa.orm.mapped_column(sa.Integer, sa.ForeignKey('team.id'), nullable=False) - team = relationship( + team_id: Mapped[int] = sa.orm.mapped_column( + sa.Integer, sa.ForeignKey('team.id'), nullable=False + ) + team: Mapped[Team] = relationship( Team, foreign_keys=[team_id], backref=backref('client_permissions', cascade='all'), ) #: AuthClient app they are assigned on - auth_client_id = sa.orm.mapped_column( + auth_client_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('auth_client.id'), nullable=False, index=True ) auth_client: Mapped[AuthClient] = with_roles( @@ -734,7 +740,7 @@ class AuthClientTeamPermissions(BaseMixin, Model): grants_via={None: {'owner'}}, ) #: The permissions as a string of tokens - access_permissions = sa.orm.mapped_column( + access_permissions: Mapped[str] = sa.orm.mapped_column( 'permissions', sa.UnicodeText, default='', nullable=False ) @@ -759,7 +765,7 @@ def get( @classmethod def all_for( cls, auth_client: AuthClient, account: Account - ) -> Query[AuthClientPermissions]: + ) -> Query[AuthClientTeamPermissions]: """Get all permissions for the specified account via their teams.""" return cls.query.filter( cls.auth_client == auth_client, diff --git a/funnel/models/comment.py b/funnel/models/comment.py index e6f3e4240..87d11a475 100644 --- a/funnel/models/comment.py +++ b/funnel/models/comment.py @@ -89,7 +89,7 @@ class SET_TYPE: # noqa: N801 class Commentset(UuidMixin, BaseMixin, Model): __tablename__ = 'commentset' #: Commentset state code - _state = sa.orm.mapped_column( + _state: Mapped[int] = sa.orm.mapped_column( 'state', sa.SmallInteger, StateManager.check_constraint('state', COMMENTSET_STATE), @@ -105,7 +105,7 @@ class Commentset(UuidMixin, BaseMixin, Model): datasets={'primary'}, ) #: Count of comments, stored to avoid count(*) queries - count = with_roles( + count: Mapped[int] = with_roles( sa.orm.mapped_column(sa.Integer, default=0, nullable=False), read={'all'}, datasets={'primary'}, @@ -214,7 +214,7 @@ class Comment(UuidMixin, BaseMixin, Model): ), grants={'author'}, ) - commentset_id = sa.orm.mapped_column( + commentset_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('commentset.id'), nullable=False ) commentset: Mapped[Commentset] = with_roles( @@ -225,7 +225,7 @@ class Comment(UuidMixin, BaseMixin, Model): grants_via={None: {'document_subscriber'}}, ) - in_reply_to_id = sa.orm.mapped_column( + in_reply_to_id: Mapped[int | None] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('comment.id'), nullable=True ) replies: Mapped[list[Comment]] = relationship( @@ -236,7 +236,7 @@ class Comment(UuidMixin, BaseMixin, Model): 'message', nullable=False ) - _state = sa.orm.mapped_column( + _state: Mapped[int] = sa.orm.mapped_column( 'state', sa.Integer, StateManager.check_constraint('state', COMMENT_STATE), @@ -245,18 +245,18 @@ class Comment(UuidMixin, BaseMixin, Model): ) state = StateManager('_state', COMMENT_STATE, doc="Current state of the comment") - edited_at = with_roles( + edited_at: Mapped[datetime | None] = with_roles( sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True), read={'all'}, datasets={'primary', 'related', 'json'}, ) #: Revision number maintained by SQLAlchemy, starting at 1 - revisionid = with_roles( + revisionid: Mapped[int] = with_roles( sa.orm.mapped_column(sa.Integer, nullable=False), read={'all'} ) - search_vector: Mapped[TSVectorType] = sa.orm.mapped_column( + search_vector: Mapped[str] = sa.orm.mapped_column( TSVectorType( 'message_text', weights={'message_text': 'A'}, diff --git a/funnel/models/commentset_membership.py b/funnel/models/commentset_membership.py index 4becbd934..27cf4541e 100644 --- a/funnel/models/commentset_membership.py +++ b/funnel/models/commentset_membership.py @@ -2,6 +2,8 @@ from __future__ import annotations +from datetime import datetime + from werkzeug.utils import cached_property from coaster.sqlalchemy import DynamicAssociationProxy, with_roles @@ -58,9 +60,11 @@ class CommentsetMembership(ImmutableUserMembershipMixin, Model): parent: Mapped[Commentset] = sa.orm.synonym('commentset') #: Flag to indicate notifications are muted - is_muted = sa.orm.mapped_column(sa.Boolean, nullable=False, default=False) + is_muted: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, nullable=False, default=False + ) #: When the user visited this commentset last - last_seen_at = sa.orm.mapped_column( + last_seen_at: Mapped[datetime] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=False, default=sa.func.utcnow() ) diff --git a/funnel/models/contact_exchange.py b/funnel/models/contact_exchange.py index c9e18974b..b8eab0694 100644 --- a/funnel/models/contact_exchange.py +++ b/funnel/models/contact_exchange.py @@ -71,7 +71,7 @@ class ContactExchange(TimestampMixin, RoleMixin, Model): ), ) #: Participant whose contact was scanned - ticket_participant_id = sa.orm.mapped_column( + ticket_participant_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('ticket_participant.id', ondelete='CASCADE'), primary_key=True, @@ -82,13 +82,17 @@ class ContactExchange(TimestampMixin, RoleMixin, Model): backref=backref('scanned_contacts', passive_deletes=True), ) #: Datetime at which the scan happened - scanned_at = sa.orm.mapped_column( + scanned_at: Mapped[datetime] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=False, default=sa.func.utcnow() ) #: Note recorded by the user (plain text) - description = sa.orm.mapped_column(sa.UnicodeText, nullable=False, default='') + description: Mapped[str] = sa.orm.mapped_column( + sa.UnicodeText, nullable=False, default='' + ) #: Archived flag - archived = sa.orm.mapped_column(sa.Boolean, nullable=False, default=False) + archived: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, nullable=False, default=False + ) __roles__ = { 'owner': { diff --git a/funnel/models/email_address.py b/funnel/models/email_address.py index 986f41be2..d8d2c4771 100644 --- a/funnel/models/email_address.py +++ b/funnel/models/email_address.py @@ -4,6 +4,7 @@ import hashlib import unicodedata +from datetime import datetime from typing import TYPE_CHECKING, Any, ClassVar, Literal, cast, overload import base58 @@ -190,10 +191,12 @@ class EmailAddress(BaseMixin, Model): #: The email address, centrepiece of this model. Case preserving. #: Validated by the :func:`_validate_email` event handler - email = sa.orm.mapped_column(sa.Unicode, nullable=True) + email: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode, nullable=True) #: The domain of the email, stored for quick lookup of related addresses #: Read-only, accessible via the :property:`domain` property - _domain = sa.orm.mapped_column('domain', sa.Unicode, nullable=True, index=True) + _domain: Mapped[str | None] = sa.orm.mapped_column( + 'domain', sa.Unicode, nullable=True, index=True + ) # email_normalized is defined below @@ -220,7 +223,7 @@ class EmailAddress(BaseMixin, Model): ) #: Does this email address work? Records last known delivery state - _delivery_state = sa.orm.mapped_column( + _delivery_state: Mapped[int] = sa.orm.mapped_column( 'delivery_state', sa.Integer, StateManager.check_constraint( @@ -237,18 +240,20 @@ class EmailAddress(BaseMixin, Model): doc="Last known delivery state of this email address", ) #: Timestamp of last known delivery state - delivery_state_at = sa.orm.mapped_column( + delivery_state_at: Mapped[datetime] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=False, default=sa.func.utcnow() ) #: Timestamp of last known recipient activity resulting from sent mail - active_at = sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True) + active_at: Mapped[datetime | None] = sa.orm.mapped_column( + sa.TIMESTAMP(timezone=True), nullable=True + ) #: Is this email address blocked from being used? If so, :attr:`email` should be #: null. Blocks apply to the canonical address (without the +sub-address variation), #: so a test for whether an address is blocked should use blake2b160_canonical to #: load the record. Other records with the same canonical hash _may_ exist without #: setting the flag due to a lack of database-side enforcement - _is_blocked = sa.orm.mapped_column( + _is_blocked: Mapped[bool] = sa.orm.mapped_column( 'is_blocked', sa.Boolean, nullable=False, default=False ) diff --git a/funnel/models/geoname.py b/funnel/models/geoname.py index b4726675f..e107a497a 100644 --- a/funnel/models/geoname.py +++ b/funnel/models/geoname.py @@ -4,6 +4,7 @@ import re from collections.abc import Collection +from datetime import date from decimal import Decimal from typing import cast @@ -104,13 +105,13 @@ class GeoAdmin1Code(BaseMixin, GeonameModel): backref='has_admin1code', viewonly=True, ) - title = sa.orm.mapped_column(sa.Unicode) - ascii_title = sa.orm.mapped_column(sa.Unicode) - country_id = sa.orm.mapped_column( + title: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + ascii_title: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + country_id: Mapped[str | None] = sa.orm.mapped_column( 'country', sa.CHAR(2), sa.ForeignKey('geo_country_info.iso_alpha2') ) country: Mapped[GeoCountryInfo | None] = relationship('GeoCountryInfo') - admin1_code = sa.orm.mapped_column(sa.Unicode) + admin1_code: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) def __repr__(self) -> str: """Return representation.""" @@ -130,14 +131,14 @@ class GeoAdmin2Code(BaseMixin, GeonameModel): backref='has_admin2code', viewonly=True, ) - title = sa.orm.mapped_column(sa.Unicode) - ascii_title = sa.orm.mapped_column(sa.Unicode) - country_id = sa.orm.mapped_column( + title: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + ascii_title: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + country_id: Mapped[str | None] = sa.orm.mapped_column( 'country', sa.CHAR(2), sa.ForeignKey('geo_country_info.iso_alpha2') ) country: Mapped[GeoCountryInfo | None] = relationship('GeoCountryInfo') - admin1_code = sa.orm.mapped_column(sa.Unicode) - admin2_code = sa.orm.mapped_column(sa.Unicode) + admin1_code: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + admin2_code: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) def __repr__(self) -> str: """Return representation.""" @@ -150,17 +151,17 @@ class GeoName(BaseNameMixin, GeonameModel): __tablename__ = 'geo_name' geonameid: Mapped[int] = sa.orm.synonym('id') - ascii_title = sa.orm.mapped_column(sa.Unicode) - latitude = sa.orm.mapped_column(sa.Numeric) - longitude = sa.orm.mapped_column(sa.Numeric) - fclass = sa.orm.mapped_column(sa.CHAR(1)) - fcode = sa.orm.mapped_column(sa.Unicode) - country_id = sa.orm.mapped_column( + ascii_title: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + latitude: Mapped[Decimal | None] = sa.orm.mapped_column(sa.Numeric) + longitude: Mapped[Decimal | None] = sa.orm.mapped_column(sa.Numeric) + fclass: Mapped[str | None] = sa.orm.mapped_column(sa.CHAR(1)) + fcode: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + country_id: Mapped[str | None] = sa.orm.mapped_column( 'country', sa.CHAR(2), sa.ForeignKey('geo_country_info.iso_alpha2') ) country: Mapped[GeoCountryInfo | None] = relationship('GeoCountryInfo') - cc2 = sa.orm.mapped_column(sa.Unicode) - admin1 = sa.orm.mapped_column(sa.Unicode) + cc2: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + admin1: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) admin1_ref: Mapped[GeoAdmin1Code | None] = relationship( 'GeoAdmin1Code', uselist=False, @@ -168,14 +169,14 @@ class GeoName(BaseNameMixin, GeonameModel): 'GeoName.admin1 == foreign(GeoAdmin1Code.admin1_code))', viewonly=True, ) - admin1_id = sa.orm.mapped_column( + admin1_id: Mapped[int | None] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('geo_admin1_code.id'), nullable=True ) admin1code: Mapped[GeoAdmin1Code | None] = relationship( 'GeoAdmin1Code', uselist=False, foreign_keys=[admin1_id] ) - admin2 = sa.orm.mapped_column(sa.Unicode) + admin2: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) admin2_ref: Mapped[GeoAdmin2Code | None] = relationship( 'GeoAdmin2Code', uselist=False, @@ -184,20 +185,21 @@ class GeoName(BaseNameMixin, GeonameModel): 'GeoName.admin2 == foreign(GeoAdmin2Code.admin2_code))', viewonly=True, ) - admin2_id = sa.orm.mapped_column( + admin2_id: Mapped[int | None] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('geo_admin2_code.id'), nullable=True ) admin2code: Mapped[GeoAdmin2Code | None] = relationship( 'GeoAdmin2Code', uselist=False, foreign_keys=[admin2_id] ) - admin4 = sa.orm.mapped_column(sa.Unicode) - admin3 = sa.orm.mapped_column(sa.Unicode) - population = sa.orm.mapped_column(sa.BigInteger) - elevation = sa.orm.mapped_column(sa.Integer) - dem = sa.orm.mapped_column(sa.Integer) # Digital Elevation Model - timezone = sa.orm.mapped_column(sa.Unicode) - moddate = sa.orm.mapped_column(sa.Date) + admin4: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + admin3: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + population: Mapped[int | None] = sa.orm.mapped_column(sa.BigInteger) + elevation: Mapped[int | None] = sa.orm.mapped_column(sa.Integer) + #: Digital Elevation Model + dem: Mapped[int | None] = sa.orm.mapped_column(sa.Integer) + timezone: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode) + moddate: Mapped[date | None] = sa.orm.mapped_column(sa.Date) __table_args__ = ( sa.Index( @@ -219,11 +221,19 @@ def short_title(self) -> str: return self.has_country.title if self.has_admin1code: return ( - self.admin1code.title if self.admin1code else self.admin1_ref.title + self.admin1code.title + if self.admin1code + else self.admin1_ref.title + if self.admin1_ref + else '' ) or '' if self.has_admin2code: return ( - self.admin2code.title if self.admin2code else self.admin2_ref.title + self.admin2code.title + if self.admin2code + else self.admin2_ref.title + if self.admin2_ref + else '' ) or '' return self.ascii_title or self.title @@ -275,7 +285,7 @@ def geoname(self) -> GeoName: @property def use_title(self) -> str: - """Return a recommended usable title.""" + """Return a recommended usable title (English-only).""" usetitle = self.ascii_title or '' if self.fclass == 'A' and self.fcode and self.fcode.startswith('PCL'): if 'of the' in usetitle: @@ -326,21 +336,26 @@ def __repr__(self) -> str: def related_geonames(self) -> dict[str, GeoName]: """Return related geonames based on superior hierarchy (country, state, etc).""" - related = {} + related: dict[str, GeoName] = {} if self.admin2code and self.admin2code.geonameid != self.geonameid: related['admin2'] = self.admin2code.geoname if self.admin1code and self.admin1code.geonameid != self.geonameid: related['admin1'] = self.admin1code.geoname - if self.country and self.country.geonameid != self.geonameid: + if ( + self.country + and self.country.geonameid != self.geonameid + and self.country.geoname + ): related['country'] = self.country.geoname if ( (self.fclass, self.fcode) != ('L', 'CONT') and self.country and self.country.continent ): - related['continent'] = GeoName.query.get( - continent_codes[self.country.continent] - ) + continent = GeoName.query.get(continent_codes[self.country.continent]) + if continent: + related['continent'] = continent + return related def as_dict(self, related=True, alternate_titles=True) -> dict: @@ -581,19 +596,21 @@ class GeoAltName(BaseMixin, GeonameModel): __tablename__ = 'geo_alt_name' - geonameid = sa.orm.mapped_column( + geonameid: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('geo_name.id'), nullable=False ) geoname: Mapped[GeoName] = relationship( GeoName, backref=backref('alternate_titles', cascade='all, delete-orphan'), ) - lang = sa.orm.mapped_column(sa.Unicode, nullable=True, index=True) - title = sa.orm.mapped_column(sa.Unicode, nullable=False) - is_preferred_name = sa.orm.mapped_column(sa.Boolean, nullable=False) - is_short_name = sa.orm.mapped_column(sa.Boolean, nullable=False) - is_colloquial = sa.orm.mapped_column(sa.Boolean, nullable=False) - is_historic = sa.orm.mapped_column(sa.Boolean, nullable=False) + lang: Mapped[str | None] = sa.orm.mapped_column( + sa.Unicode, nullable=True, index=True + ) + title: Mapped[str] = sa.orm.mapped_column(sa.Unicode, nullable=False) + is_preferred_name: Mapped[str] = sa.orm.mapped_column(sa.Boolean, nullable=False) + is_short_name: Mapped[bool] = sa.orm.mapped_column(sa.Boolean, nullable=False) + is_colloquial: Mapped[bool] = sa.orm.mapped_column(sa.Boolean, nullable=False) + is_historic: Mapped[bool] = sa.orm.mapped_column(sa.Boolean, nullable=False) __table_args__ = ( sa.Index( diff --git a/funnel/models/helpers.py b/funnel/models/helpers.py index 3fb9d038e..9a38014e5 100644 --- a/funnel/models/helpers.py +++ b/funnel/models/helpers.py @@ -391,7 +391,7 @@ def add_search_trigger(model: type[Model], column_name: str) -> dict[str, str]: class MyModel(Model): ... - search_vector: Mapped[TSVectorType] = sa.orm.mapped_column( + search_vector: Mapped[str] = sa.orm.mapped_column( TSVectorType( 'name', 'title', *indexed_columns, weights={'name': 'A', 'title': 'B'}, diff --git a/funnel/models/label.py b/funnel/models/label.py index cb586b160..16cfca13b 100644 --- a/funnel/models/label.py +++ b/funnel/models/label.py @@ -45,7 +45,7 @@ class Label(BaseScopedNameMixin, Model): __tablename__ = 'label' - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id', ondelete='CASCADE'), nullable=False ) # Backref from project is defined in the Project model with an ordering list @@ -59,7 +59,7 @@ class Label(BaseScopedNameMixin, Model): #: Parent label's id. Do not write to this column directly, as we don't have the #: ability to : validate the value within the app. Always use the :attr:`main_label` #: relationship. - main_label_id = sa.orm.mapped_column( + main_label_id: Mapped[int | None] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('label.id', ondelete='CASCADE'), index=True, @@ -81,37 +81,39 @@ class Label(BaseScopedNameMixin, Model): # add_primary_relationship) #: Sequence number for this label, used in UI for ordering - seq = sa.orm.mapped_column(sa.Integer, nullable=False) + seq: Mapped[int] = sa.orm.mapped_column(sa.Integer, nullable=False) # A single-line description of this label, shown when picking labels (optional) - description = sa.orm.mapped_column(sa.UnicodeText, nullable=False, default='') + description: Mapped[str] = sa.orm.mapped_column( + sa.UnicodeText, nullable=False, default='' + ) #: Icon for displaying in space-constrained UI. Contains one emoji symbol. #: Since emoji can be composed from multiple symbols, there is no length #: limit imposed here - icon_emoji = sa.orm.mapped_column(sa.UnicodeText, nullable=True) + icon_emoji: Mapped[str | None] = sa.orm.mapped_column(sa.UnicodeText, nullable=True) #: Restricted mode specifies that this label may only be applied by someone with #: an editorial role (TODO: name the role). If this label is a parent, it applies #: to all its children - _restricted = sa.orm.mapped_column( + _restricted: Mapped[bool] = sa.orm.mapped_column( 'restricted', sa.Boolean, nullable=False, default=False ) #: Required mode signals to UI that if this label is a parent, one of its #: children must be mandatorily applied to the proposal. The value of this #: field must be ignored if the label is not a parent - _required = sa.orm.mapped_column( + _required: Mapped[bool] = sa.orm.mapped_column( 'required', sa.Boolean, nullable=False, default=False ) #: Archived mode specifies that the label is no longer available for use #: although all the previous records will stay in database. - _archived = sa.orm.mapped_column( + _archived: Mapped[bool] = sa.orm.mapped_column( 'archived', sa.Boolean, nullable=False, default=False ) - search_vector: Mapped[TSVectorType] = sa.orm.mapped_column( + search_vector: Mapped[str] = sa.orm.mapped_column( TSVectorType( 'name', 'title', diff --git a/funnel/models/login_session.py b/funnel/models/login_session.py index e1d628cc5..73c8bee53 100644 --- a/funnel/models/login_session.py +++ b/funnel/models/login_session.py @@ -2,7 +2,7 @@ from __future__ import annotations -from datetime import timedelta +from datetime import datetime, timedelta from coaster.utils import utcnow @@ -95,23 +95,31 @@ class LoginSession(UuidMixin, BaseMixin, Model): ) #: User's last known IP address - ipaddr = sa.orm.mapped_column(sa.String(45), nullable=False) + ipaddr: Mapped[str] = sa.orm.mapped_column(sa.String(45), nullable=False) #: City geonameid from IP address - geonameid_city = sa.orm.mapped_column(sa.Integer, nullable=True) + geonameid_city: Mapped[int | None] = sa.orm.mapped_column(sa.Integer, nullable=True) #: State/subdivision geonameid from IP address - geonameid_subdivision = sa.orm.mapped_column(sa.Integer, nullable=True) + geonameid_subdivision: Mapped[int | None] = sa.orm.mapped_column( + sa.Integer, nullable=True + ) #: Country geonameid from IP address - geonameid_country = sa.orm.mapped_column(sa.Integer, nullable=True) + geonameid_country: Mapped[int | None] = sa.orm.mapped_column( + sa.Integer, nullable=True + ) #: User's network, from IP address - geoip_asn = sa.orm.mapped_column(sa.Integer, nullable=True) + geoip_asn: Mapped[int | None] = sa.orm.mapped_column(sa.Integer, nullable=True) #: User agent - user_agent = sa.orm.mapped_column(sa.UnicodeText, nullable=False) + user_agent: Mapped[str] = sa.orm.mapped_column(sa.UnicodeText, nullable=False) #: The login service that was used to make this session - login_service = sa.orm.mapped_column(sa.Unicode, nullable=True) + login_service: Mapped[str | None] = sa.orm.mapped_column(sa.Unicode, nullable=True) - accessed_at = sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=False) - revoked_at = sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True) - sudo_enabled_at = sa.orm.mapped_column( + accessed_at: Mapped[datetime] = sa.orm.mapped_column( + sa.TIMESTAMP(timezone=True), nullable=False + ) + revoked_at: Mapped[datetime | None] = sa.orm.mapped_column( + sa.TIMESTAMP(timezone=True), nullable=True + ) + sudo_enabled_at: Mapped[datetime] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=False, default=sa.func.utcnow() ) diff --git a/funnel/models/membership_mixin.py b/funnel/models/membership_mixin.py index 37e61d46b..eb918dd8a 100644 --- a/funnel/models/membership_mixin.py +++ b/funnel/models/membership_mixin.py @@ -399,8 +399,13 @@ def user(cls) -> Mapped[Account]: @classmethod def __table_args__(cls) -> tuple: """Table arguments for SQLAlchemy.""" + try: + args = list(super().__table_args__) # type: ignore[misc] + except AttributeError: + args = [] + kwargs = args.pop(-1) if args and isinstance(args[-1], dict) else None if cls.parent_id_column is not None: - return ( + args.append( sa.Index( 'ix_' + cls.__tablename__ + '_active', cls.parent_id_column, @@ -409,14 +414,18 @@ def __table_args__(cls) -> tuple: postgresql_where='revoked_at IS NULL', ), ) - return ( - sa.Index( - 'ix_' + cls.__tablename__ + '_active', - 'member_id', - unique=True, - postgresql_where='revoked_at IS NULL', - ), - ) + else: + args.append( + sa.Index( + 'ix_' + cls.__tablename__ + '_active', + 'member_id', + unique=True, + postgresql_where='revoked_at IS NULL', + ), + ) + if kwargs: + args.append(kwargs) + return tuple(args) @hybrid_property def is_self_granted(self) -> bool: @@ -513,7 +522,11 @@ def seq(cls) -> Mapped[int]: @classmethod def __table_args__(cls) -> tuple: """Table arguments.""" - args = list(super().__table_args__) # type: ignore[misc] + try: + args = list(super().__table_args__) # type: ignore[misc] + except AttributeError: + args = [] + kwargs = args.pop(-1) if args and isinstance(args[-1], dict) else None # Add unique constraint on :attr:`seq` for active records args.append( sa.Index( @@ -524,6 +537,8 @@ def __table_args__(cls) -> tuple: postgresql_where='revoked_at IS NULL', ), ) + if kwargs: + args.append(kwargs) return tuple(args) def __init__(self, **kwargs) -> None: @@ -582,7 +597,8 @@ def _title(cls) -> Mapped[str | None]: def title(self) -> str: """Attribution title for this record.""" if self._local_data_only: - return self._title # This may be None # type: ignore[return-value] + # self._title may be None + return self._title # type: ignore[return-value] return self._title or self.member.title @title.setter diff --git a/funnel/models/moderation.py b/funnel/models/moderation.py index 7ec499771..52b2e73f6 100644 --- a/funnel/models/moderation.py +++ b/funnel/models/moderation.py @@ -2,6 +2,7 @@ from __future__ import annotations +from datetime import datetime from uuid import UUID from baseframe import __ @@ -25,7 +26,7 @@ class MODERATOR_REPORT_TYPE(LabeledEnum): # noqa: N801 class CommentModeratorReport(UuidMixin, BaseMixin[UUID], Model): __tablename__ = 'comment_moderator_report' - comment_id = sa.orm.mapped_column( + comment_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('comment.id'), nullable=False, index=True ) comment: Mapped[Comment] = relationship( @@ -41,16 +42,16 @@ class CommentModeratorReport(UuidMixin, BaseMixin[UUID], Model): foreign_keys=[reported_by_id], backref=backref('moderator_reports', cascade='all', lazy='dynamic'), ) - report_type = sa.orm.mapped_column( + report_type: Mapped[int] = sa.orm.mapped_column( sa.SmallInteger, StateManager.check_constraint('report_type', MODERATOR_REPORT_TYPE), nullable=False, default=MODERATOR_REPORT_TYPE.SPAM, ) - reported_at = sa.orm.mapped_column( + reported_at: Mapped[datetime] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), default=sa.func.utcnow(), nullable=False ) - resolved_at = sa.orm.mapped_column( + resolved_at: Mapped[datetime | None] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=True, index=True ) diff --git a/funnel/models/notification.py b/funnel/models/notification.py index 21a8c17dd..62936cb85 100644 --- a/funnel/models/notification.py +++ b/funnel/models/notification.py @@ -1258,7 +1258,7 @@ class NotificationPreferences(BaseMixin, Model): index=True, ) #: User account whose preferences are represented here - account = with_roles( + account: Mapped[Account] = with_roles( relationship(Account, back_populates='notification_preferences'), read={'owner'}, grants={'owner'}, diff --git a/funnel/models/phone_number.py b/funnel/models/phone_number.py index f49c7515b..02b8135a4 100644 --- a/funnel/models/phone_number.py +++ b/funnel/models/phone_number.py @@ -9,8 +9,7 @@ import base58 import phonenumbers from sqlalchemy import event, inspect -from sqlalchemy.orm import Mapper -from sqlalchemy.orm.attributes import NO_VALUE +from sqlalchemy.orm import NO_VALUE, Mapper from sqlalchemy.sql.expression import ColumnElement from werkzeug.utils import cached_property diff --git a/funnel/models/project.py b/funnel/models/project.py index 04353bcf3..a889c4369 100644 --- a/funnel/models/project.py +++ b/funnel/models/project.py @@ -3,8 +3,10 @@ from __future__ import annotations from collections.abc import Sequence +from datetime import datetime -from pytz import utc +from furl import furl +from pytz import BaseTzInfo, utc from sqlalchemy.orm import attribute_keyed_dict from werkzeug.utils import cached_property @@ -71,12 +73,16 @@ class Project(UuidMixin, BaseScopedNameMixin, Model): __tablename__ = 'project' reserved_names = RESERVED_NAMES - created_by_id = sa.orm.mapped_column(sa.ForeignKey('account.id'), nullable=False) + created_by_id: Mapped[int] = sa.orm.mapped_column( + sa.ForeignKey('account.id'), nullable=False + ) created_by: Mapped[Account] = relationship( Account, foreign_keys=[created_by_id], ) - account_id = sa.orm.mapped_column(sa.ForeignKey('account.id'), nullable=False) + account_id: Mapped[int] = sa.orm.mapped_column( + sa.ForeignKey('account.id'), nullable=False + ) account: Mapped[Account] = with_roles( relationship( Account, @@ -113,25 +119,25 @@ class Project(UuidMixin, BaseScopedNameMixin, Model): ) = MarkdownCompositeDocument.create('instructions', default='', nullable=True) with_roles(instructions, read={'all'}) - location = with_roles( + location: Mapped[str | None] = with_roles( sa.orm.mapped_column(sa.Unicode(50), default='', nullable=True), read={'all'}, datasets={'primary', 'without_parent', 'related'}, ) parsed_location: Mapped[types.jsonb_dict] - website = with_roles( + website: Mapped[furl | None] = with_roles( sa.orm.mapped_column(UrlType, nullable=True), read={'all'}, datasets={'primary', 'without_parent'}, ) - timezone = with_roles( + timezone: Mapped[BaseTzInfo] = with_roles( sa.orm.mapped_column(TimezoneType(backend='pytz'), nullable=False, default=utc), read={'all'}, datasets={'primary', 'without_parent', 'related'}, ) - _state = sa.orm.mapped_column( + _state: Mapped[int] = sa.orm.mapped_column( 'state', sa.Integer, StateManager.check_constraint('state', PROJECT_STATE), @@ -142,7 +148,7 @@ class Project(UuidMixin, BaseScopedNameMixin, Model): state = with_roles( StateManager('_state', PROJECT_STATE, doc="Project state"), call={'all'} ) - _cfp_state = sa.orm.mapped_column( + _cfp_state: Mapped[int] = sa.orm.mapped_column( 'cfp_state', sa.Integer, StateManager.check_constraint('cfp_state', CFP_STATE), @@ -155,39 +161,39 @@ class Project(UuidMixin, BaseScopedNameMixin, Model): ) #: Audit timestamp to detect re-publishing to re-surface a project - first_published_at = sa.orm.mapped_column( + first_published_at: Mapped[datetime | None] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=True ) #: Timestamp of when this project was most recently published - published_at = with_roles( + published_at: Mapped[datetime | None] = with_roles( sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True, index=True), read={'all'}, write={'promoter'}, datasets={'primary', 'without_parent', 'related'}, ) #: Optional start time for schedule, cached from column property schedule_start_at - start_at = with_roles( + start_at: Mapped[datetime | None] = with_roles( sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True, index=True), read={'all'}, write={'editor'}, datasets={'primary', 'without_parent', 'related'}, ) #: Optional end time for schedule, cached from column property schedule_end_at - end_at = with_roles( + end_at: Mapped[datetime | None] = with_roles( sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True, index=True), read={'all'}, write={'editor'}, datasets={'primary', 'without_parent', 'related'}, ) - cfp_start_at = sa.orm.mapped_column( + cfp_start_at: Mapped[datetime | None] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=True, index=True ) - cfp_end_at = sa.orm.mapped_column( + cfp_end_at: Mapped[datetime | None] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=True, index=True ) - bg_image = with_roles( + bg_image: Mapped[furl | None] = with_roles( sa.orm.mapped_column(ImgeeType, nullable=True), read={'all'}, datasets={'primary', 'without_parent', 'related'}, @@ -203,13 +209,13 @@ class Project(UuidMixin, BaseScopedNameMixin, Model): read={'all'}, datasets={'primary', 'without_parent', 'related'}, ) - buy_tickets_url: Mapped[str | None] = with_roles( + buy_tickets_url: Mapped[furl | None] = with_roles( sa.orm.mapped_column(UrlType, nullable=True), read={'all'}, datasets={'primary', 'without_parent', 'related'}, ) - banner_video_url = with_roles( + banner_video_url: Mapped[furl | None] = with_roles( sa.orm.mapped_column(UrlType, nullable=True), read={'all'}, datasets={'primary', 'without_parent'}, @@ -223,15 +229,15 @@ class Project(UuidMixin, BaseScopedNameMixin, Model): call={'all'}, ) - hasjob_embed_url = with_roles( + hasjob_embed_url: Mapped[furl | None] = with_roles( sa.orm.mapped_column(UrlType, nullable=True), read={'all'} ) - hasjob_embed_limit = with_roles( - sa.orm.mapped_column(sa.Integer, default=8), read={'all'} + hasjob_embed_limit: Mapped[int | None] = with_roles( + sa.orm.mapped_column(sa.Integer, default=8, nullable=True), read={'all'} ) - commentset_id = sa.orm.mapped_column( - sa.Integer, sa.ForeignKey('commentset.id'), nullable=False + commentset_id: Mapped[int] = sa.orm.mapped_column( + sa.ForeignKey('commentset.id'), nullable=False ) commentset: Mapped[Commentset] = relationship( Commentset, @@ -241,25 +247,27 @@ class Project(UuidMixin, BaseScopedNameMixin, Model): back_populates='project', ) - parent_id = sa.orm.mapped_column( - sa.Integer, sa.ForeignKey('project.id', ondelete='SET NULL'), nullable=True + parent_id: Mapped[int | None] = sa.orm.mapped_column( + sa.ForeignKey('project.id', ondelete='SET NULL'), nullable=True ) parent_project: Mapped[Project | None] = relationship( - 'Project', remote_side='Project.id', backref='subprojects' + remote_side='Project.id', back_populates='subprojects' ) + subprojects: Mapped[list[Project]] = relationship(back_populates='parent_project') #: Featured project flag. This can only be set by website editors, not #: project editors or account admins. - site_featured = with_roles( + site_featured: Mapped[bool] = with_roles( sa.orm.mapped_column(sa.Boolean, default=False, nullable=False), read={'all'}, write={'site_editor'}, datasets={'primary', 'without_parent'}, ) - livestream_urls = with_roles( + livestream_urls: Mapped[list[str] | None] = with_roles( sa.orm.mapped_column( sa.ARRAY(sa.UnicodeText, dimensions=1), + nullable=True, # For legacy data server_default=sa.text("'{}'::text[]"), ), read={'all'}, @@ -273,11 +281,11 @@ class Project(UuidMixin, BaseScopedNameMixin, Model): ) #: Revision number maintained by SQLAlchemy, used for vCal files, starting at 1 - revisionid = with_roles( + revisionid: Mapped[int] = with_roles( sa.orm.mapped_column(sa.Integer, nullable=False), read={'all'} ) - search_vector: Mapped[TSVectorType] = sa.orm.mapped_column( + search_vector: Mapped[str] = sa.orm.mapped_column( TSVectorType( 'name', 'title', @@ -498,7 +506,7 @@ def __format__(self, format_spec: str) -> str: message=__("Submissions will be accepted until the optional closing date"), type='success', ) - def open_cfp(self): + def open_cfp(self) -> None: """Change state to accept submissions.""" # If closing date is in the past, remove it if self.cfp_end_at is not None and self.cfp_end_at <= utcnow(): @@ -515,7 +523,7 @@ def open_cfp(self): message=__("Submissions will no longer be accepted"), type='success', ) - def close_cfp(self): + def close_cfp(self) -> None: """Change state to not accept submissions.""" @with_roles(call={'editor'}) @@ -800,7 +808,7 @@ class __Account: ), viewonly=True, ) - projects_by_name = with_roles( + projects_by_name: Mapped[dict[str, Project]] = with_roles( relationship( Project, foreign_keys=[Project.account_id], @@ -870,10 +878,10 @@ class ProjectRedirect(TimestampMixin, Model): sa.Unicode(250), nullable=False, primary_key=True ) - project_id = sa.orm.mapped_column( + project_id: Mapped[int | None] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id', ondelete='SET NULL'), nullable=True ) - project: Mapped[Project] = relationship(Project, backref='redirects') + project: Mapped[Project | None] = relationship(Project, backref='redirects') def __repr__(self) -> str: """Represent :class:`ProjectRedirect` as a string.""" @@ -941,17 +949,19 @@ def migrate_account(cls, old_account: Account, new_account: Account) -> None: class ProjectLocation(TimestampMixin, Model): __tablename__ = 'project_location' #: Project we are tagging - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id'), primary_key=True, nullable=False ) project: Mapped[Project] = relationship( Project, backref=backref('locations', cascade='all') ) #: Geonameid for this project - geonameid = sa.orm.mapped_column( + geonameid: Mapped[int] = sa.orm.mapped_column( sa.Integer, primary_key=True, nullable=False, index=True ) - primary = sa.orm.mapped_column(sa.Boolean, default=True, nullable=False) + primary: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, default=True, nullable=False + ) def __repr__(self) -> str: """Represent :class:`ProjectLocation` as a string.""" @@ -963,7 +973,7 @@ def __repr__(self) -> str: @reopen(Commentset) class __Commentset: - project = with_roles( + project: Mapped[Project | None] = with_roles( relationship(Project, uselist=False, back_populates='commentset'), grants_via={None: {'editor': 'document_subscriber'}}, ) diff --git a/funnel/models/project_membership.py b/funnel/models/project_membership.py index fda29e6b3..07b36f6f2 100644 --- a/funnel/models/project_membership.py +++ b/funnel/models/project_membership.py @@ -155,7 +155,10 @@ class ProjectMembership(ImmutableUserMembershipMixin, Model): @classmethod def __table_args__(cls) -> tuple: """Table arguments.""" - args = list(super().__table_args__) + try: + args = list(super().__table_args__) # type: ignore[misc] + except AttributeError: + args = [] kwargs = args.pop(-1) if args and isinstance(args[-1], dict) else None args.append( sa.CheckConstraint( diff --git a/funnel/models/proposal.py b/funnel/models/proposal.py index 3b79396de..1124e3e51 100644 --- a/funnel/models/proposal.py +++ b/funnel/models/proposal.py @@ -123,8 +123,10 @@ class Proposal( # type: ignore[misc] ): __tablename__ = 'proposal' - created_by_id = sa.orm.mapped_column(sa.ForeignKey('account.id'), nullable=False) - created_by = with_roles( + created_by_id: Mapped[int] = sa.orm.mapped_column( + sa.ForeignKey('account.id'), nullable=False + ) + created_by: Mapped[Account] = with_roles( relationship( Account, foreign_keys=[created_by_id], @@ -132,7 +134,7 @@ class Proposal( # type: ignore[misc] ), grants={'creator', 'participant'}, ) - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id'), nullable=False ) project: Mapped[Project] = with_roles( @@ -162,7 +164,7 @@ class Proposal( # type: ignore[misc] # TODO: Stand-in for `submitted_at` until proposals have a workflow-driven datetime datetime: Mapped[datetime_type] = sa.orm.synonym('created_at') - _state = sa.orm.mapped_column( + _state: Mapped[int] = sa.orm.mapped_column( 'state', sa.Integer, StateManager.check_constraint('state', PROPOSAL_STATE), @@ -171,7 +173,7 @@ class Proposal( # type: ignore[misc] ) state = StateManager('_state', PROPOSAL_STATE, doc="Current state of the proposal") - commentset_id = sa.orm.mapped_column( + commentset_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('commentset.id'), nullable=False ) commentset: Mapped[Commentset] = relationship( @@ -186,19 +188,29 @@ class Proposal( # type: ignore[misc] body, body_text, body_html = MarkdownCompositeDocument.create( 'body', nullable=False, default='' ) - description = sa.orm.mapped_column(sa.Unicode, nullable=False, default='') - custom_description = sa.orm.mapped_column(sa.Boolean, nullable=False, default=False) - template = sa.orm.mapped_column(sa.Boolean, nullable=False, default=False) - featured = sa.orm.mapped_column(sa.Boolean, nullable=False, default=False) + description: Mapped[str] = sa.orm.mapped_column( + sa.Unicode, nullable=False, default='' + ) + custom_description: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, nullable=False, default=False + ) + template: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, nullable=False, default=False + ) + featured: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, nullable=False, default=False + ) - edited_at = sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True) + edited_at: Mapped[datetime_type | None] = sa.orm.mapped_column( + sa.TIMESTAMP(timezone=True), nullable=True + ) #: Revision number maintained by SQLAlchemy, starting at 1 - revisionid = with_roles( + revisionid: Mapped[int] = with_roles( sa.orm.mapped_column(sa.Integer, nullable=False), read={'all'} ) - search_vector: Mapped[TSVectorType] = sa.orm.mapped_column( + search_vector: Mapped[str] = sa.orm.mapped_column( TSVectorType( 'title', 'description', @@ -513,8 +525,10 @@ class ProposalSuuidRedirect(BaseMixin, Model): __tablename__ = 'proposal_suuid_redirect' - suuid = sa.orm.mapped_column(sa.Unicode(22), nullable=False, index=True) - proposal_id = sa.orm.mapped_column( + suuid: Mapped[str] = sa.orm.mapped_column( + sa.Unicode(22), nullable=False, index=True + ) + proposal_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('proposal.id', ondelete='CASCADE'), nullable=False ) proposal: Mapped[Proposal] = relationship(Proposal) diff --git a/funnel/models/proposal_membership.py b/funnel/models/proposal_membership.py index 115d086bc..fba987e46 100644 --- a/funnel/models/proposal_membership.py +++ b/funnel/models/proposal_membership.py @@ -106,7 +106,9 @@ class ProposalMembership( # type: ignore[misc] #: Uncredited members are not listed in the main display, but can edit and may be #: listed in a details section. Uncredited memberships are for support roles such #: as copy editors. - is_uncredited = sa.orm.mapped_column(sa.Boolean, nullable=False, default=False) + is_uncredited: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, nullable=False, default=False + ) #: Optional label, indicating the member's role on the proposal label = immutable( @@ -131,7 +133,7 @@ class __Proposal: # This relationship does not use `lazy='dynamic'` because it is expected to contain # <2 records on average, and won't exceed 50 in the most extreme cases - memberships = with_roles( + memberships: Mapped[list[ProposalMembership]] = with_roles( relationship( ProposalMembership, primaryjoin=sa.and_( diff --git a/funnel/models/rsvp.py b/funnel/models/rsvp.py index ac6390686..66026e9a5 100644 --- a/funnel/models/rsvp.py +++ b/funnel/models/rsvp.py @@ -42,7 +42,7 @@ class RSVP_STATUS(LabeledEnum): # noqa: N801 class Rsvp(UuidMixin, NoIdMixin, Model): __tablename__ = 'rsvp' - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id'), nullable=False, primary_key=True ) project: Mapped[Project] = with_roles( @@ -67,7 +67,7 @@ class Rsvp(UuidMixin, NoIdMixin, Model): datasets={'primary', 'without_parent', 'related'}, ) - _state = sa.orm.mapped_column( + _state: Mapped[str] = sa.orm.mapped_column( 'state', sa.CHAR(1), StateManager.check_constraint('state', RSVP_STATUS), diff --git a/funnel/models/saved.py b/funnel/models/saved.py index 4310c14ad..48c9fb13a 100644 --- a/funnel/models/saved.py +++ b/funnel/models/saved.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Sequence +from datetime import datetime from coaster.sqlalchemy import LazyRoleSet, with_roles @@ -27,7 +28,7 @@ class SavedProject(NoIdMixin, Model): backref=backref('saved_projects', lazy='dynamic', passive_deletes=True), ) #: Project that was saved - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id', ondelete='CASCADE'), nullable=False, @@ -39,11 +40,13 @@ class SavedProject(NoIdMixin, Model): backref=backref('saved_by', lazy='dynamic', passive_deletes=True), ) #: Timestamp when the save happened - saved_at = sa.orm.mapped_column( + saved_at: Mapped[datetime] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=False, default=sa.func.utcnow() ) #: User's plaintext note to self on why they saved this (optional) - description = sa.orm.mapped_column(sa.UnicodeText, nullable=True) + description: Mapped[str | None] = sa.orm.mapped_column( + sa.UnicodeText, nullable=True + ) def roles_for( self, actor: Account | None = None, anchors: Sequence = () @@ -78,7 +81,7 @@ class SavedSession(NoIdMixin, Model): backref=backref('saved_sessions', lazy='dynamic', passive_deletes=True), ) #: Session that was saved - session_id = sa.orm.mapped_column( + session_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('session.id', ondelete='CASCADE'), nullable=False, @@ -90,11 +93,13 @@ class SavedSession(NoIdMixin, Model): backref=backref('saved_by', lazy='dynamic', passive_deletes=True), ) #: Timestamp when the save happened - saved_at = sa.orm.mapped_column( + saved_at: Mapped[datetime] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=False, default=sa.func.utcnow() ) #: User's plaintext note to self on why they saved this (optional) - description = sa.orm.mapped_column(sa.UnicodeText, nullable=True) + description: Mapped[str | None] = sa.orm.mapped_column( + sa.UnicodeText, nullable=True + ) def roles_for( self, actor: Account | None = None, anchors: Sequence = () diff --git a/funnel/models/session.py b/funnel/models/session.py index f6b34289c..1d1b9fdf1 100644 --- a/funnel/models/session.py +++ b/funnel/models/session.py @@ -67,20 +67,26 @@ class Session(UuidMixin, BaseScopedIdNameMixin, VideoMixin, Model): proposal: Mapped[Proposal | None] = relationship( Proposal, backref=backref('session', uselist=False, cascade='all') ) - speaker = sa.orm.mapped_column(sa.Unicode(200), default=None, nullable=True) - start_at = sa.orm.mapped_column( + speaker: Mapped[str | None] = sa.orm.mapped_column( + sa.Unicode(200), default=None, nullable=True + ) + start_at: Mapped[datetime | None] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=True, index=True ) - end_at = sa.orm.mapped_column( + end_at: Mapped[datetime | None] = sa.orm.mapped_column( sa.TIMESTAMP(timezone=True), nullable=True, index=True ) - venue_room_id = sa.orm.mapped_column( + venue_room_id: Mapped[int | None] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('venue_room.id'), nullable=True ) venue_room: Mapped[VenueRoom | None] = relationship(VenueRoom, backref='sessions') - is_break = sa.orm.mapped_column(sa.Boolean, default=False, nullable=False) - featured = sa.orm.mapped_column(sa.Boolean, default=False, nullable=False) - is_restricted_video = sa.orm.mapped_column( + is_break: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, default=False, nullable=False + ) + featured: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, default=False, nullable=False + ) + is_restricted_video: Mapped[bool] = sa.orm.mapped_column( sa.Boolean, default=False, nullable=False ) banner_image_url: Mapped[str | None] = sa.orm.mapped_column( @@ -88,11 +94,11 @@ class Session(UuidMixin, BaseScopedIdNameMixin, VideoMixin, Model): ) #: Version number maintained by SQLAlchemy, used for vCal files, starting at 1 - revisionid = with_roles( + revisionid: Mapped[int] = with_roles( sa.orm.mapped_column(sa.Integer, nullable=False), read={'all'} ) - search_vector: Mapped[TSVectorType] = sa.orm.mapped_column( + search_vector: Mapped[str] = sa.orm.mapped_column( TSVectorType( 'title', 'description_text', diff --git a/funnel/models/shortlink.py b/funnel/models/shortlink.py index fb04c4795..a30cecdd3 100644 --- a/funnel/models/shortlink.py +++ b/funnel/models/shortlink.py @@ -197,7 +197,7 @@ class Shortlink(NoIdMixin, Model): is_new = False # id of this shortlink, saved as a bigint (8 bytes) - id = with_roles( # noqa: A003 + id: Mapped[int] = with_roles( # noqa: A003 # id cannot use the `immutable` wrapper because :meth:`new` changes the id when # handling collisions. This needs an "immutable after commit" handler sa.orm.mapped_column( @@ -206,7 +206,7 @@ class Shortlink(NoIdMixin, Model): read={'all'}, ) #: URL target of this shortlink - url = with_roles( + url: Mapped[furl] = with_roles( immutable(sa.orm.mapped_column(UrlType, nullable=False, index=True)), read={'all'}, ) @@ -218,7 +218,9 @@ class Shortlink(NoIdMixin, Model): created_by: Mapped[Account | None] = relationship(Account) #: Is this link enabled? If not, render 410 Gone - enabled = sa.orm.mapped_column(sa.Boolean, nullable=False, default=True) + enabled: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, nullable=False, default=True + ) @hybrid_property def name(self) -> str: diff --git a/funnel/models/site_membership.py b/funnel/models/site_membership.py index 2121e13dc..c1167435f 100644 --- a/funnel/models/site_membership.py +++ b/funnel/models/site_membership.py @@ -66,7 +66,10 @@ class SiteMembership(ImmutableUserMembershipMixin, Model): @classmethod def __table_args__(cls) -> tuple: """Table arguments.""" - args = list(super().__table_args__) + try: + args = list(super().__table_args__) # type: ignore[misc] + except AttributeError: + args = [] args.append( sa.CheckConstraint( sa.or_( diff --git a/funnel/models/sync_ticket.py b/funnel/models/sync_ticket.py index 924570375..cef344296 100644 --- a/funnel/models/sync_ticket.py +++ b/funnel/models/sync_ticket.py @@ -114,7 +114,7 @@ class TicketEvent(GetTitleMixin, Model): __tablename__ = 'ticket_event' - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id'), nullable=False ) project: Mapped[Project] = with_roles( @@ -140,7 +140,7 @@ class TicketEvent(GetTitleMixin, Model): ), rw={'project_promoter'}, ) - badge_template = with_roles( + badge_template: Mapped[str | None] = with_roles( sa.orm.mapped_column(sa.Unicode(250), nullable=True), rw={'project_promoter'} ) @@ -172,7 +172,7 @@ class TicketType(GetTitleMixin, Model): __tablename__ = 'ticket_type' - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id'), nullable=False ) project: Mapped[Project] = with_roles( @@ -213,50 +213,52 @@ class TicketParticipant(EmailAddressMixin, UuidMixin, BaseMixin, Model): __email_optional__ = True __email_for__ = 'participant' - fullname = with_roles( + fullname: Mapped[str] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=False), read={'promoter', 'member', 'scanner'}, ) #: Unvalidated phone number - phone = with_roles( + phone: Mapped[str | None] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=True), read={'promoter', 'member', 'scanner'}, ) #: Unvalidated Twitter id - twitter = with_roles( + twitter: Mapped[str | None] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=True), read={'promoter', 'member', 'scanner'}, ) #: Job title - job_title = with_roles( + job_title: Mapped[str | None] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=True), read={'promoter', 'member', 'scanner'}, ) #: Company - company = with_roles( + company: Mapped[str | None] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=True), read={'promoter', 'member', 'scanner'}, ) #: Participant's city - city = with_roles( + city: Mapped[str | None] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=True), read={'promoter', 'member', 'scanner'}, ) # public key - puk = sa.orm.mapped_column( + puk: Mapped[str] = sa.orm.mapped_column( sa.Unicode(44), nullable=False, default=make_public_key, unique=True ) - key = sa.orm.mapped_column( + key: Mapped[str] = sa.orm.mapped_column( sa.Unicode(44), nullable=False, default=make_private_key, unique=True ) - badge_printed = sa.orm.mapped_column(sa.Boolean, default=False, nullable=False) + badge_printed: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, default=False, nullable=False + ) participant_id: Mapped[int | None] = sa.orm.mapped_column( sa.ForeignKey('account.id'), nullable=True ) participant: Mapped[Account | None] = relationship( Account, backref=backref('ticket_participants', cascade='all') ) - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id'), nullable=False ) project: Mapped[Project] = with_roles( @@ -396,7 +398,7 @@ class TicketEventParticipant(BaseMixin, Model): __tablename__ = 'ticket_event_participant' - ticket_participant_id = sa.orm.mapped_column( + ticket_participant_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('ticket_participant.id'), nullable=False ) ticket_participant: Mapped[TicketParticipant] = relationship( @@ -408,7 +410,7 @@ class TicketEventParticipant(BaseMixin, Model): ), overlaps='ticket_events,ticket_participants', ) - ticket_event_id = sa.orm.mapped_column( + ticket_event_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('ticket_event.id'), nullable=False ) ticket_event: Mapped[TicketEvent] = relationship( @@ -420,7 +422,9 @@ class TicketEventParticipant(BaseMixin, Model): ), overlaps='ticket_events,ticket_participants', ) - checked_in = sa.orm.mapped_column(sa.Boolean, default=False, nullable=False) + checked_in: Mapped[bool] = sa.orm.mapped_column( + sa.Boolean, default=False, nullable=False + ) __table_args__ = ( # Uses a custom name that is not as per convention because the default name is @@ -448,25 +452,25 @@ def get( class TicketClient(BaseMixin, Model): __tablename__ = 'ticket_client' - name = with_roles( + name: Mapped[str] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'} ) - client_eventid = with_roles( + client_eventid: Mapped[str] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'} ) - clientid = with_roles( + clientid: Mapped[str] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'} ) - client_secret = with_roles( + client_secret: Mapped[str] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'} ) - client_access_token = with_roles( + client_access_token: Mapped[str] = with_roles( sa.orm.mapped_column(sa.Unicode(80), nullable=False), rw={'project_promoter'} ) - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id'), nullable=False ) - project = with_roles( + project: Mapped[Project] = with_roles( relationship(Project, backref=backref('ticket_clients', cascade='all')), rw={'project_promoter'}, grants_via={None: project_child_role_map}, @@ -520,22 +524,22 @@ class SyncTicket(BaseMixin, Model): __tablename__ = 'sync_ticket' - ticket_no = sa.orm.mapped_column(sa.Unicode(80), nullable=False) - order_no = sa.orm.mapped_column(sa.Unicode(80), nullable=False) - ticket_type_id = sa.orm.mapped_column( + ticket_no: Mapped[str] = sa.orm.mapped_column(sa.Unicode(80), nullable=False) + order_no: Mapped[str] = sa.orm.mapped_column(sa.Unicode(80), nullable=False) + ticket_type_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('ticket_type.id'), nullable=False ) ticket_type: Mapped[TicketType] = relationship( TicketType, backref=backref('sync_tickets', cascade='all') ) - ticket_participant_id = sa.orm.mapped_column( + ticket_participant_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('ticket_participant.id'), nullable=False ) ticket_participant: Mapped[TicketParticipant] = relationship( TicketParticipant, backref=backref('sync_tickets', cascade='all'), ) - ticket_client_id = sa.orm.mapped_column( + ticket_client_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('ticket_client.id'), nullable=False ) ticket_client: Mapped[TicketClient] = relationship( diff --git a/funnel/models/update.py b/funnel/models/update.py index 3f637aeb2..f579a386f 100644 --- a/funnel/models/update.py +++ b/funnel/models/update.py @@ -3,6 +3,7 @@ from __future__ import annotations from collections.abc import Sequence +from datetime import datetime from sqlalchemy.orm import Query as BaseQuery @@ -49,7 +50,7 @@ class VISIBILITY_STATE(LabeledEnum): # noqa: N801 class Update(UuidMixin, BaseScopedIdNameMixin, Model): __tablename__ = 'update' - _visibility_state = sa.orm.mapped_column( + _visibility_state: Mapped[int] = sa.orm.mapped_column( 'visibility_state', sa.SmallInteger, StateManager.check_constraint('visibility_state', VISIBILITY_STATE), @@ -61,7 +62,7 @@ class Update(UuidMixin, BaseScopedIdNameMixin, Model): '_visibility_state', VISIBILITY_STATE, doc="Visibility state" ) - _state = sa.orm.mapped_column( + _state: Mapped[int] = sa.orm.mapped_column( 'state', sa.SmallInteger, StateManager.check_constraint('state', UPDATE_STATE), @@ -84,7 +85,7 @@ class Update(UuidMixin, BaseScopedIdNameMixin, Model): grants={'creator'}, ) - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id'), nullable=False, index=True ) project: Mapped[Project] = with_roles( @@ -125,13 +126,13 @@ class Update(UuidMixin, BaseScopedIdNameMixin, Model): ) #: Update number, for Project updates, assigned when the update is published - number = with_roles( + number: Mapped[int | None] = with_roles( sa.orm.mapped_column(sa.Integer, nullable=True, default=None), read={'all'} ) #: Like pinned tweets. You can keep posting updates, #: but might want to pin an update from a week ago. - is_pinned = with_roles( + is_pinned: Mapped[bool] = with_roles( sa.orm.mapped_column(sa.Boolean, default=False, nullable=False), read={'all'} ) @@ -146,7 +147,7 @@ class Update(UuidMixin, BaseScopedIdNameMixin, Model): ), read={'all'}, ) - published_at = with_roles( + published_at: Mapped[datetime | None] = with_roles( sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True), read={'all'} ) @@ -161,19 +162,19 @@ class Update(UuidMixin, BaseScopedIdNameMixin, Model): ), read={'reader'}, ) - deleted_at = with_roles( + deleted_at: Mapped[datetime | None] = with_roles( sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True), read={'reader'}, ) - edited_at = with_roles( + edited_at: Mapped[datetime | None] = with_roles( sa.orm.mapped_column(sa.TIMESTAMP(timezone=True), nullable=True), read={'all'} ) - commentset_id = sa.orm.mapped_column( + commentset_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('commentset.id'), nullable=False ) - commentset = with_roles( + commentset: Mapped[Commentset] = with_roles( relationship( Commentset, uselist=False, @@ -185,7 +186,7 @@ class Update(UuidMixin, BaseScopedIdNameMixin, Model): read={'all'}, ) - search_vector: Mapped[TSVectorType] = sa.orm.mapped_column( + search_vector: Mapped[str] = sa.orm.mapped_column( TSVectorType( 'name', 'title', diff --git a/funnel/models/venue.py b/funnel/models/venue.py index 620da0b01..585dbd07b 100644 --- a/funnel/models/venue.py +++ b/funnel/models/venue.py @@ -25,7 +25,7 @@ class Venue(UuidMixin, BaseScopedNameMixin, CoordinatesMixin, Model): __tablename__ = 'venue' - project_id = sa.orm.mapped_column( + project_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('project.id'), nullable=False ) project: Mapped[Project] = with_roles( @@ -36,12 +36,22 @@ class Venue(UuidMixin, BaseScopedNameMixin, CoordinatesMixin, Model): description, description_text, description_html = MarkdownCompositeBasic.create( 'description', default='', nullable=False ) - address1 = sa.orm.mapped_column(sa.Unicode(160), default='', nullable=False) - address2 = sa.orm.mapped_column(sa.Unicode(160), default='', nullable=False) - city = sa.orm.mapped_column(sa.Unicode(30), default='', nullable=False) - state = sa.orm.mapped_column(sa.Unicode(30), default='', nullable=False) - postcode = sa.orm.mapped_column(sa.Unicode(20), default='', nullable=False) - country = sa.orm.mapped_column(sa.Unicode(2), default='', nullable=False) + address1: Mapped[str] = sa.orm.mapped_column( + sa.Unicode(160), default='', nullable=False + ) + address2: Mapped[str] = sa.orm.mapped_column( + sa.Unicode(160), default='', nullable=False + ) + city: Mapped[str] = sa.orm.mapped_column(sa.Unicode(30), default='', nullable=False) + state: Mapped[str] = sa.orm.mapped_column( + sa.Unicode(30), default='', nullable=False + ) + postcode: Mapped[str] = sa.orm.mapped_column( + sa.Unicode(20), default='', nullable=False + ) + country: Mapped[str] = sa.orm.mapped_column( + sa.Unicode(2), default='', nullable=False + ) rooms: Mapped[list[VenueRoom]] = relationship( 'VenueRoom', @@ -51,7 +61,7 @@ class Venue(UuidMixin, BaseScopedNameMixin, CoordinatesMixin, Model): back_populates='venue', ) - seq = sa.orm.mapped_column(sa.Integer, nullable=False) + seq: Mapped[int] = sa.orm.mapped_column(sa.Integer, nullable=False) __table_args__ = (sa.UniqueConstraint('project_id', 'name'),) @@ -105,7 +115,7 @@ class Venue(UuidMixin, BaseScopedNameMixin, CoordinatesMixin, Model): class VenueRoom(UuidMixin, BaseScopedNameMixin, Model): __tablename__ = 'venue_room' - venue_id = sa.orm.mapped_column( + venue_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('venue.id'), nullable=False ) venue: Mapped[Venue] = with_roles( @@ -117,9 +127,11 @@ class VenueRoom(UuidMixin, BaseScopedNameMixin, Model): description, description_text, description_html = MarkdownCompositeBasic.create( 'description', default='', nullable=False ) - bgcolor = sa.orm.mapped_column(sa.Unicode(6), nullable=False, default='229922') + bgcolor: Mapped[str] = sa.orm.mapped_column( + sa.Unicode(6), nullable=False, default='229922' + ) - seq = sa.orm.mapped_column(sa.Integer, nullable=False) + seq: Mapped[int] = sa.orm.mapped_column(sa.Integer, nullable=False) __table_args__ = (sa.UniqueConstraint('venue_id', 'name'),) @@ -174,7 +186,7 @@ def scoped_name(self): @reopen(Project) class __Project: - venues = with_roles( + venues: Mapped[list[Venue]] = with_roles( relationship( Venue, cascade='all', diff --git a/funnel/models/video_mixin.py b/funnel/models/video_mixin.py index f8816dca7..1775d2490 100644 --- a/funnel/models/video_mixin.py +++ b/funnel/models/video_mixin.py @@ -4,7 +4,7 @@ from furl import furl -from . import declarative_mixin, sa +from . import Mapped, declarative_mixin, sa __all__ = ['VideoMixin', 'VideoError', 'parse_video_url'] @@ -14,8 +14,9 @@ class VideoError(Exception): def parse_video_url(video_url: str) -> tuple[str, str]: + video_id: str | None + video_id = video_url video_source = 'raw' - video_id: str | None = video_url parsed = furl(video_url) if not parsed.host: @@ -74,8 +75,10 @@ def make_video_url(video_source: str, video_id: str) -> str: @declarative_mixin class VideoMixin: - video_id = sa.orm.mapped_column(sa.UnicodeText, nullable=True) - video_source = sa.orm.mapped_column(sa.UnicodeText, nullable=True) + video_id: Mapped[str | None] = sa.orm.mapped_column(sa.UnicodeText, nullable=True) + video_source: Mapped[str | None] = sa.orm.mapped_column( + sa.UnicodeText, nullable=True + ) @property def video_url(self) -> str | None: diff --git a/migrations/script.py.mako b/migrations/script.py.mako index 88b3d025d..f132556e3 100644 --- a/migrations/script.py.mako +++ b/migrations/script.py.mako @@ -17,7 +17,7 @@ ${imports if imports else ""} revision: str = ${repr(up_revision)} down_revision: str = ${repr(down_revision)} branch_labels: str | tuple[str, ...] | None = ${repr(branch_labels)} -depends_on: str, tuple[str, ...] | None = ${repr(depends_on)} +depends_on: str | tuple[str, ...] | None = ${repr(depends_on)} def upgrade(engine_name: str = '') -> None: diff --git a/tests/unit/models/email_address_test.py b/tests/unit/models/email_address_test.py index c3f552f10..38dab0594 100644 --- a/tests/unit/models/email_address_test.py +++ b/tests/unit/models/email_address_test.py @@ -8,8 +8,10 @@ import pytest import sqlalchemy as sa from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Mapped from funnel import models +from funnel.models import relationship # This hash map should not be edited -- hashes are permanent hash_map = { @@ -422,10 +424,10 @@ class EmailLink(models.EmailAddressMixin, models.BaseMixin, models.Model): __email_for__ = 'emailuser' __email_is_exclusive__ = True - emailuser_id = sa.orm.mapped_column( + emailuser_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('test_email_user.id'), nullable=False ) - emailuser = models.relationship(EmailUser) + emailuser = relationship(EmailUser) class EmailDocument(models.EmailAddressMixin, models.BaseMixin, models.Model): """Test model unaffiliated to a user that has an email address attached.""" @@ -438,10 +440,10 @@ class EmailLinkedDocument(models.EmailAddressMixin, models.BaseMixin, models.Mod __tablename__ = 'test_email_linked_document' __email_for__ = 'emailuser' - emailuser_id = sa.orm.mapped_column( + emailuser_id: Mapped[int | None] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('test_email_user.id'), nullable=True ) - emailuser = models.relationship(EmailUser) + emailuser: Mapped[EmailUser | None] = relationship(EmailUser) new_models = [EmailUser, EmailLink, EmailDocument, EmailLinkedDocument] diff --git a/tests/unit/models/helpers_test.py b/tests/unit/models/helpers_test.py index 4ff1821be..b0b7dc3b0 100644 --- a/tests/unit/models/helpers_test.py +++ b/tests/unit/models/helpers_test.py @@ -6,7 +6,9 @@ import pytest import sqlalchemy as sa from flask_babel import lazy_gettext +from furl import furl from sqlalchemy.exc import StatementError +from sqlalchemy.orm import Mapped import funnel.models.helpers as mhelpers from funnel import models @@ -168,8 +170,10 @@ def new_foobar(self): # skipcq: PTC-W0049 def image_models(database, app): class MyImageModel(models.Model): __tablename__ = 'test_my_image_model' - id = sa.orm.mapped_column(sa.Integer, primary_key=True) # noqa: A003 - image_url = sa.orm.mapped_column(models.ImgeeType) + id: Mapped[int] = sa.orm.mapped_column( # noqa: A003 + sa.Integer, primary_key=True + ) + image_url: Mapped[furl] = sa.orm.mapped_column(models.ImgeeType) with app.app_context(): database.create_all() diff --git a/tests/unit/models/phone_number_test.py b/tests/unit/models/phone_number_test.py index f33269f3b..1945dc8a1 100644 --- a/tests/unit/models/phone_number_test.py +++ b/tests/unit/models/phone_number_test.py @@ -9,8 +9,10 @@ import pytest import sqlalchemy as sa from sqlalchemy.exc import IntegrityError +from sqlalchemy.orm import Mapped from funnel import models +from funnel.models import relationship # These numbers were obtained from libphonenumber with region codes 'IN' and 'US': # >>> phonenumbers.example_number_for_type(region, phonenumbers.PhoneNumberType.MOBILE) @@ -59,10 +61,10 @@ class PhoneLink(models.PhoneNumberMixin, models.BaseMixin, models.Model): __phone_for__ = 'phoneuser' __phone_is_exclusive__ = True - phoneuser_id = sa.orm.mapped_column( + phoneuser_id: Mapped[int] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('test_phone_user.id'), nullable=False ) - phoneuser = models.relationship(PhoneUser) + phoneuser: Mapped[PhoneUser] = relationship(PhoneUser) class PhoneDocument(models.PhoneNumberMixin, models.BaseMixin, models.Model): """Test model unaffiliated to a user that has a phone number attached.""" @@ -75,10 +77,10 @@ class PhoneLinkedDocument(models.PhoneNumberMixin, models.BaseMixin, models.Mode __tablename__ = 'test_phone_linked_document' __phone_for__ = 'phoneuser' - phoneuser_id = sa.orm.mapped_column( + phoneuser_id: Mapped[int | None] = sa.orm.mapped_column( sa.Integer, sa.ForeignKey('test_phone_user.id'), nullable=True ) - phoneuser = models.relationship(PhoneUser) + phoneuser: Mapped[PhoneUser | None] = relationship(PhoneUser) new_models = [PhoneUser, PhoneLink, PhoneDocument, PhoneLinkedDocument]