diff --git a/pycroft/model/__init__.py b/pycroft/model/__init__.py index 2c6c15463..c84d8b95d 100644 --- a/pycroft/model/__init__.py +++ b/pycroft/model/__init__.py @@ -12,7 +12,7 @@ from datetime import timezone, tzinfo import psycopg2.extensions -from sqlalchemy import create_engine as sqa_create_engine +from sqlalchemy import create_engine as sqa_create_engine, Connection from sqlalchemy.future import Engine from . import _all @@ -61,7 +61,7 @@ def create_engine(connection_string, **kwargs) -> Engine: return sqa_create_engine(connection_string, **kwargs) -def create_db_model(bind): +def create_db_model(bind: Connection) -> None: """Create all models in the database. """ # skip objects marked with "is_view" @@ -70,7 +70,7 @@ def create_db_model(bind): base.ModelBase.metadata.create_all(bind, tables=tables) -def drop_db_model(bind): +def drop_db_model(bind: Connection) -> None: """Drop all models from the database. """ # skip objects marked with "is_view" diff --git a/pycroft/model/session.py b/pycroft/model/session.py index 9f42928ba..0d7e51a94 100644 --- a/pycroft/model/session.py +++ b/pycroft/model/session.py @@ -13,6 +13,7 @@ from typing import overload, TypeVar, Callable, Any, TYPE_CHECKING from sqlalchemy.orm import scoped_session +from sqlalchemy.sql.functions import AnsiFunction from werkzeug.local import LocalProxy import wrapt @@ -77,3 +78,7 @@ def with_transaction(wrapped, instance, args, kwargs): def utcnow() -> DateTimeTz: return session.query(func.current_timestamp()).scalar() + + +def current_timestamp() -> AnsiFunction[DateTimeTz]: + return t.cast(AnsiFunction[DateTimeTz], func.current_timestamp()) diff --git a/pycroft/model/user.py b/pycroft/model/user.py index 827b39bcb..50000588d 100644 --- a/pycroft/model/user.py +++ b/pycroft/model/user.py @@ -185,14 +185,15 @@ def validate_passwd_hash(self, _, value): "not correct!" return value - def check_password(self, plaintext_password): + def check_password(self, plaintext_password: str) -> bool: """verify a given plaintext password against the users passwd hash. """ return verify_password(plaintext_password, self.passwd_hash) @property - def password(self): + # actually `NoReturn`, but mismatch to `setter` confuses mypy + def password(self) -> str: """Store a hash of a given plaintext passwd for the user.""" raise RuntimeError("Password can not be read, only set") diff --git a/stubs/flask_restful/reqparse.pyi b/stubs/flask_restful/reqparse.pyi index 492877e06..e2a23a70e 100644 --- a/stubs/flask_restful/reqparse.pyi +++ b/stubs/flask_restful/reqparse.pyi @@ -35,7 +35,7 @@ class RequestParser: trim: Incomplete bundle_errors: Incomplete def __init__(self, argument_class=..., namespace_class=..., trim: bool = ..., bundle_errors: bool = ...) -> None: ... - def add_argument(self, *args, **kwargs): ... + def add_argument(self, *args, **kwargs) -> RequestParser: ... def parse_args(self, req: Incomplete | None = ..., strict: bool = ..., http_error_code: int = ...): ... def copy(self): ... def replace_argument(self, name, *args, **kwargs): ... diff --git a/web/api/v0/__init__.py b/web/api/v0/__init__.py index c2ae05aa6..efd782f5b 100644 --- a/web/api/v0/__init__.py +++ b/web/api/v0/__init__.py @@ -8,7 +8,7 @@ from flask_restful import Api, Resource as FlaskRestfulResource, abort, \ reqparse, inputs from sqlalchemy.exc import IntegrityError -from sqlalchemy import select, func +from sqlalchemy import select from sqlalchemy.orm import joinedload, selectinload from pycroft.helpers import utc @@ -33,6 +33,7 @@ from pycroft.model.facilities import Room from pycroft.model.finance import Account, Split from pycroft.model.host import IP, Interface, Host +from pycroft.model.session import current_timestamp from pycroft.model.types import IPAddress, InvalidMACAddressException from pycroft.model.user import User, IllegalEmailError, IllegalLoginError from web.api.v0.helpers import parse_iso_date @@ -105,10 +106,8 @@ def generate_user_data(user: User) -> Response: step = timedelta(days=1) traffic_history = func_traffic_history( user.id, - # TODO what is the emitted sql statement? - # it seems to me that this expression returns `timestamp`, and not `timestamptz` - func.current_timestamp() - interval + step, - func.current_timestamp(), + current_timestamp() - interval + step, + current_timestamp(), ) class _Entry(t.TypedDict): diff --git a/web/blueprints/finance/__init__.py b/web/blueprints/finance/__init__.py index daff02296..c6a52b960 100644 --- a/web/blueprints/finance/__init__.py +++ b/web/blueprints/finance/__init__.py @@ -736,7 +736,7 @@ def balance_json(account_id: int) -> ResponseReturnValue: sum_exp: ColumnElement[int] = t.cast( Over[int], - func.sum(Split.amount).over(order_by=Transaction.valid_on), + func.sum(Split.amount).over(order_by=Transaction.valid_on), # type: ignore[no-untyped-call] ) if invert: diff --git a/web/blueprints/user/tables.py b/web/blueprints/user/tables.py index a80ecb5fc..66880b8a0 100644 --- a/web/blueprints/user/tables.py +++ b/web/blueprints/user/tables.py @@ -1,5 +1,4 @@ import typing as t -import typing from flask import url_for from pydantic import BaseModel diff --git a/web/commands.py b/web/commands.py index 50e723799..96bda96bb 100644 --- a/web/commands.py +++ b/web/commands.py @@ -1,6 +1,7 @@ # Copyright (c) 2021. The Pycroft Authors. See the AUTHORS file. # This file is part of the Pycroft project and licensed under the terms of # the Apache License, Version 2.0. See the LICENSE file for details +import typing as t import os import click @@ -10,17 +11,19 @@ from pycroft.model import create_engine, drop_db_model -def register_commands(app: Flask): +def register_commands(app: Flask) -> None: """Register custom commands executable via `flask $command_name`.""" - @app.cli.command('create-model', help="Create the database model.") - def create_model(): + cli = t.cast(click.Group, app.cli) + + @cli.command("create-model", help="Create the database model.") + def create_model() -> None: engine = create_engine(os.getenv('PYCROFT_DB_URI')) with engine.begin() as connection: create_db_model(bind=connection) - @app.cli.command('drop-model', help="Drop the database model.") - def drop_model(): + @cli.command("drop-model", help="Drop the database model.") + def drop_model() -> None: engine = create_engine(os.getenv('PYCROFT_DB_URI')) click.confirm(f'This will drop the whole database schema associated to {engine!r}.' ' Are you absolutely sure?', abort=True) diff --git a/web/form/widgets.py b/web/form/widgets.py index ad2b765c5..461750a62 100644 --- a/web/form/widgets.py +++ b/web/form/widgets.py @@ -1,3 +1,6 @@ +import typing as t + + import wtforms_widgets from wtforms import ValidationError, Form @@ -8,10 +11,10 @@ class UserIDField(wtforms_widgets.fields.core.StringField): """A User-ID Field """ - def __init__(self, *args, **kwargs) -> None: + def __init__(self, *args: t.Any, **kwargs: t.Any) -> None: super().__init__(*args, **kwargs) - def __call__(self, **kwargs) -> None: + def __call__(self, **kwargs: t.Any) -> t.Any: return super().__call__( **kwargs ) diff --git a/web/template_filters.py b/web/template_filters.py index 65f062390..c64226e1c 100644 --- a/web/template_filters.py +++ b/web/template_filters.py @@ -9,24 +9,31 @@ :copyright: (c) 2012 by AG DSN. """ +import typing as t import pathlib from cmath import log +from datetime import datetime, date +from decimal import Decimal from itertools import chain from re import sub import flask_babel -from flask import current_app, json, url_for +from flask import current_app, json, url_for, Flask from jinja2 import pass_context from jinja2.runtime import Context from pycroft.helpers.i18n import localized, gettext +from pycroft.helpers.utc import ensure_tz from pycroft.model import session _filter_registry = {} -def template_filter(name): - def decorator(fn): +_TF = t.TypeVar("_TF", bound=t.Callable[..., t.Any]) + + +def template_filter(name: str) -> t.Callable[[_TF], _TF]: + def decorator(fn: _TF) -> _TF: _filter_registry[name] = fn return fn return decorator @@ -48,7 +55,7 @@ class AssetNotFound(Exception): # noinspection PyUnusedLocal @template_filter("require") @pass_context -def require(ctx: Context, asset: str, **kwargs) -> str: +def require(ctx: Context, asset: str, **kwargs: t.Any) -> str: """ Build an URL for an asset generated by webpack. @@ -74,7 +81,7 @@ def require(ctx: Context, asset: str, **kwargs) -> str: with path.open() as f: asset_map = json.load(f) - def has_changed(): + def has_changed() -> bool: try: return path.stat().st_mtime != mtime except OSError: @@ -86,47 +93,45 @@ def has_changed(): filename = asset_map[asset] except KeyError: raise AssetNotFound(f"Asset {asset} not found") from None - kwargs['filename'] = filename - return url_for('static', **kwargs) + kwargs["filename"] = filename + return url_for("static", **kwargs) @template_filter("pretty_category") -def pretty_category_filter(category): +def pretty_category_filter(category: str) -> str: """Make pretty category names for flash messages, etc """ return _category_map.get(category, "Hinweis") @template_filter("date") -def date_filter(dt, format=None): +def date_filter(dt: datetime | date | None, format: str | None = None) -> str: """Format date or datetime objects using Flask-Babel - :param datetime|date|None dt: a datetime object or None - :param str format: format as understood by Flask-Babel's format_datetime - :rtype: unicode + :param dt: a datetime object or None + :param format: format as understood by Flask-Babel's format_datetime """ if dt is None: return "k/A" - return flask_babel.format_date(dt, format) + return t.cast(str, flask_babel.format_date(dt, format)) @template_filter("datetime") -def datetime_filter(dt, format=None): +def datetime_filter(dt: datetime | None, format: str | None = None) -> str: """Format datetime objects using Flask-Babel - :param datetime|None dt: a datetime object or None - :param str format: format as understood by Flask-Babel's format_datetime - :rtype: unicode + :param dt: a datetime object or None + :param format: format as understood by Flask-Babel's format_datetime """ if dt is None: return "k/A" if isinstance(dt, str): return dt - return flask_babel.format_datetime(dt, format) + return t.cast(str, flask_babel.format_datetime(dt, format)) @template_filter("timesince") -def timesince_filter(dt, default="just now"): +def timesince_filter(dt: datetime | None, default: str = "just now") -> str: """ Returns string representing "time since" e.g. 3 days ago, 5 hours ago etc. @@ -138,7 +143,7 @@ def timesince_filter(dt, default="just now"): return "k/A" now = session.utcnow() - diff = now - dt + diff = now - ensure_tz(dt) periods = ( (diff.days / 365, "Jahr", "Jahre"), @@ -149,16 +154,18 @@ def timesince_filter(dt, default="just now"): (diff.seconds / 60, "Minute", "Minuten"), (diff.seconds, "Sekunde", "Sekunden"), ) - - for period, singular, plural in periods: - - if period: - return f"vor {period:d} {singular if period == 1 else plural}" - - return default + return next( + ( + f"vor {period:d} {singular if period == 1 else plural}" + for period, singular, plural in periods + ), + default, + ) -def prefix_unit_filter(value, unit, factor, prefixes): +def prefix_unit_filter( + value: float | Decimal, unit: str, factor: int, prefixes: t.Iterable[str] +) -> str: units = list(chain(unit, (p + unit for p in prefixes))) if value > 0: n = min(int(log(value, factor).real), len(units)-1) @@ -169,17 +176,17 @@ def prefix_unit_filter(value, unit, factor, prefixes): @template_filter("byte_size") -def byte_size_filter(value): +def byte_size_filter(value: float | Decimal) -> str: return prefix_unit_filter(value, 'B', 1024, ['Ki', 'Mi', 'Gi', 'Ti']) @template_filter("money") -def money_filter(amount): +def money_filter(amount: float | Decimal) -> str: return (f"{amount:.2f}\u202f€").replace('.', ',') @template_filter("icon") -def icon_filter(icon_class: str): +def icon_filter(icon_class: str) -> str: if len(tokens := icon_class.split(maxsplit=1)) == 2: prefix, icon = tokens else: @@ -189,7 +196,7 @@ def icon_filter(icon_class: str): @template_filter("account_type") -def account_type_filter(account_type): +def account_type_filter(account_type: str) -> str: types = { "USER_ASSET": gettext("User account (asset)"), "BANK_ASSET": gettext("Bank account (asset)"), @@ -204,9 +211,14 @@ def account_type_filter(account_type): @template_filter("transaction_type") -def transaction_type_filter(credit_debit_type): - def replacer(types): - return types and tuple(sub(r'[A-Z]+_(?=ASSET)', r'', t) for t in types) +def transaction_type_filter(credit_debit_type: tuple[str, str]) -> str: + def remove_prefix(account_type_name: str) -> str: + return sub(r"[A-Z]+_(?=ASSET)", r"", account_type_name) + + def replacer(types: tuple[str, str]) -> tuple[str, str] | None: + if not types: + return None + return (remove_prefix(types[0]), remove_prefix(types[1])) types = { ("ASSET", "LIABILITY"): gettext("Balance sheet extension"), @@ -246,6 +258,6 @@ def host_traffic_filter(host): """ -def register_filters(app): +def register_filters(app: Flask) -> None: for name in _filter_registry: app.jinja_env.filters[name] = _filter_registry[name] diff --git a/web/template_tests.py b/web/template_tests.py index 4b8e498fc..efeb61e4e 100644 --- a/web/template_tests.py +++ b/web/template_tests.py @@ -1,35 +1,42 @@ # Copyright (c) 2014 The Pycroft Authors. See the AUTHORS file. # This file is part of the Pycroft project and licensed under the terms of # the Apache License, Version 2.0. See the LICENSE file for details. +import typing as t + +from flask import Flask from pycroft.lib.user import has_positive_balance +from pycroft.model.user import User + +_check_registry: dict[str, t.Callable] = {} + -_check_registry = {} +_T = t.TypeVar("_T", bound=t.Callable) -def template_check(name): - def decorator(fn): +def template_check(name: str) -> t.Callable[[_T], _T]: + def decorator(fn: _T) -> _T: _check_registry[name] = fn return fn return decorator @template_check("user_with_positive_balance") -def positive_balance_check(user): +def positive_balance_check(user: User) -> bool: """Tests if user has a positive balance """ return has_positive_balance(user) @template_check("user_with_no_network_access") -def no_network_access_check(user): +def no_network_access_check(user: User) -> bool: """Tests if user has network access """ return not user.has_property("network_access") @template_check("privileged_for") -def privilege_check(user, *required_privileges): +def privilege_check(user: User, *required_privileges: str) -> bool: """Tests if the user has one of the required_privileges to view the requested component. """ @@ -40,34 +47,34 @@ def privilege_check(user, *required_privileges): @template_check("greater") -def greater(value, other): +def greater(value: t.Any, other: t.Any) -> bool: """Tests if another value is greater than a given value.""" - return value < other + return bool(value < other) @template_check("less") -def less(value, other): +def less(value: t.Any, other: t.Any) -> bool: """Tests if another value is less than a given value.""" - return value > other + return bool(value > other) @template_check("greater_equal") -def greater_equal(value, other): +def greater_equal(value: t.Any, other: t.Any) -> bool: """Tests if another value is greater than or equal a given value.""" - return value <= other + return bool(value <= other) @template_check("less_equal") -def less_equal(value, other): +def less_equal(value: t.Any, other: t.Any) -> bool: """Tests if another value is less than or equal a given value.""" - return value >= other + return bool(value >= other) @template_check("is_dict") -def is_dict(value): +def is_dict(value: t.Any) -> bool: return isinstance(value, dict) -def register_checks(app): +def register_checks(app: Flask) -> None: for name in _check_registry: app.jinja_env.tests[name] = _check_registry[name] diff --git a/web/templates/__init__.py b/web/templates/__init__.py index a1dc59451..1ff0ccbbf 100644 --- a/web/templates/__init__.py +++ b/web/templates/__init__.py @@ -1,44 +1,55 @@ # Copyright (c) 2014 The Pycroft Authors. See the AUTHORS file. # This file is part of the Pycroft project and licensed under the terms of # the Apache License, Version 2.0. See the LICENSE file for details. -from collections import OrderedDict, namedtuple +import typing as t +from collections import OrderedDict +from flask import Flask from flask.globals import request_ctx -LinkedScript = namedtuple("LinkedScript", ("url", "mime_type")) -PageResources = namedtuple("PageResources", ("script_files", "ready_scripts", - "stylesheet_files")) +class LinkedScript(t.NamedTuple): + url: str + mime_type: str + + +class PageResources(t.NamedTuple): + script_files: OrderedDict[str, LinkedScript] + ready_scripts: list[str] + stylesheet_files: OrderedDict[str, str] class PageResourceRegistry: """Register resources like script files for later inclusion in pages.""" @property - def page_resources(self): + def page_resources(self) -> PageResources: """Page resources are attached to Flask's current request context.""" ctx = request_ctx - if not hasattr(ctx, "page_resources"): - ctx.page_resources = PageResources(OrderedDict(), list(), - OrderedDict()) - return ctx.page_resources + if hasattr(ctx, "page_resources"): + assert isinstance(ctx.page_resources, PageResources) + return ctx.page_resources + + res = PageResources(OrderedDict(), [], OrderedDict()) + ctx.page_resources = res # type: ignore[attr-defined] + return res @property - def script_files(self): + def script_files(self) -> OrderedDict[str, LinkedScript]: return self.page_resources.script_files @property - def ready_scripts(self): + def ready_scripts(self) -> list[str]: return self.page_resources.ready_scripts @property - def stylesheet_files(self): + def stylesheet_files(self) -> OrderedDict[str, str]: return self.page_resources.stylesheet_files - def link_stylesheet(self, url): + def link_stylesheet(self, url: str) -> None: """Link a stylesheet file using a URL""" self.stylesheet_files.setdefault(url, url) - def link_script(self, url, mime_type="text/javascript"): + def link_script(self, url: str, mime_type: str = "text/javascript") -> None: """ Link a script file using a URL. @@ -47,7 +58,7 @@ def link_script(self, url, mime_type="text/javascript"): """ self.script_files.setdefault(url, LinkedScript(url, mime_type)) - def append_ready_script(self, script): + def append_ready_script(self, script: str) -> None: """ Register a script as jQuery onReady handler. @@ -56,7 +67,7 @@ def append_ready_script(self, script): """ self.ready_scripts.append(script) - def init_app(self, app): + def init_app(self, app: Flask) -> None: app.context_processor(lambda: {"page_resources": self})