From 114247ca113d2a862faa14ac583f2c33294aaf8b Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Mon, 21 Aug 2023 12:16:17 +0200 Subject: [PATCH 01/18] [typing] Stricter mypy rules for `web.blueprints.host` --- pyproject.toml | 2 +- web/blueprints/helpers/form.py | 7 +++- web/blueprints/host/__init__.py | 67 +++++++++++++++++---------------- web/blueprints/host/tables.py | 10 ++--- 4 files changed, 46 insertions(+), 40 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ad389218a..09c1f4e5e 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -52,7 +52,7 @@ module = [ "web.blueprints", "web.blueprints.access", "web.blueprints.facilities", -# "web.blueprints.host", + "web.blueprints.host", "web.blueprints.navigation", ] disallow_untyped_defs = true diff --git a/web/blueprints/helpers/form.py b/web/blueprints/helpers/form.py index 8ea0f5042..26f788426 100644 --- a/web/blueprints/helpers/form.py +++ b/web/blueprints/helpers/form.py @@ -1,17 +1,22 @@ +from __future__ import annotations import typing +import typing as t from wtforms import Form from wtforms_widgets.fields.core import BooleanField from pycroft.model.facilities import Room +if t.TYPE_CHECKING: + from web.blueprints.facilities.forms import SelectRoomForm + def iter_prefixed_field_names(cls: type[Form], prefix: str) -> typing.Iterator[str]: return (f for f in cls.__dict__ if hasattr(f, '_formfield') and f.startswith(prefix)) -def refill_room_data(form, room): +def refill_room_data(form: SelectRoomForm, room: Room) -> None: if room: form.building.data = room.building diff --git a/web/blueprints/host/__init__.py b/web/blueprints/host/__init__.py index 74ab6a42d..2f633afd4 100644 --- a/web/blueprints/host/__init__.py +++ b/web/blueprints/host/__init__.py @@ -26,12 +26,16 @@ access = BlueprintAccess(bp, required_properties=['user_show']) +def get_host_or_404(host_id: int) -> Host: + if (host := session.session.get(Host, host_id)) is None: + flash("Host existiert nicht.", "error") + abort(404) + return host + @bp.route('//delete', methods=['GET', 'POST']) @access.require('hosts_change') -def host_delete(host_id) -> ResponseReturnValue: - if (host := Host.get(host_id)) is None: - flash("Host existiert nicht.", 'error') - abort(404) +def host_delete(host_id: int) -> ResponseReturnValue: + host = get_host_or_404(host_id) owner = host.owner form = FlaskForm() @@ -61,10 +65,8 @@ def default_response() -> ResponseReturnValue: @bp.route('//edit', methods=['GET', 'POST']) @access.require('hosts_change') -def host_edit(host_id) -> ResponseReturnValue: - if (host := Host.get(host_id)) is None: - flash("Host existiert nicht.", 'error') - abort(404) +def host_edit(host_id: int) -> ResponseReturnValue: + host = get_host_or_404(host_id) form = HostForm(obj=host) def default_response() -> ResponseReturnValue: @@ -85,7 +87,7 @@ def default_response() -> ResponseReturnValue: return default_response() # existence guaranteed by validator - owner = User.get(form.owner_id.data) + owner = session.session.get(User, form.owner_id.data) with abort_on_error(default_response), session.session.begin_nested(): if not ( room := get_room( @@ -129,7 +131,8 @@ def default_response() -> ResponseReturnValue: return default_response() # existence verified by validator - owner = User.get(form.owner_id.data) + # TODO can't we provide an attribute for that on the form? + owner = session.session.get(User, form.owner_id.data) with abort_on_error(default_response), session.session.begin_nested(): if not ( room := get_room( @@ -155,10 +158,8 @@ def default_response() -> ResponseReturnValue: @bp.route("//interfaces") -def host_interfaces_json(host_id) -> ResponseReturnValue: - if (host := Host.get(host_id)) is None: - flash("Host existiert nicht.", 'error') - abort(404) +def host_interfaces_json(host_id: int) -> ResponseReturnValue: + host = get_host_or_404(host_id) return TableResponse[InterfaceRow]( items=[ @@ -189,7 +190,7 @@ def host_interfaces_json(host_id) -> ResponseReturnValue: @bp.route("//interfaces/table") -def interface_table(host_id) -> ResponseReturnValue: +def interface_table(host_id: int) -> ResponseReturnValue: return render_template('host/interface_table.html', interface_table=InterfaceTable( data_url=url_for(".host_interfaces_json", @@ -198,12 +199,17 @@ def interface_table(host_id) -> ResponseReturnValue: host_id=host_id) +def get_interface_or_404(id: int) -> Interface: + if (interface := session.session.get(Interface, id)) is None: + flash("Interface existiert nicht.", "error") + abort(404) + return interface + + @bp.route('/interface//delete', methods=['GET', 'POST']) @access.require('hosts_change') -def interface_delete(interface_id) -> ResponseReturnValue: - if (interface := Interface.get(interface_id)) is None: - flash("Interface existiert nicht.", 'error') - abort(404) +def interface_delete(interface_id: int) -> ResponseReturnValue: + interface = get_interface_or_404(interface_id) form = FlaskForm() @@ -236,10 +242,8 @@ def default_response() -> ResponseReturnValue: @bp.route('/interface//edit', methods=['GET', 'POST']) @access.require('hosts_change') -def interface_edit(interface_id) -> ResponseReturnValue: - if (interface := Interface.get(interface_id)) is None: - flash("Interface existiert nicht.", 'error') - abort(404) +def interface_edit(interface_id: int) -> ResponseReturnValue: + interface = get_interface_or_404(interface_id) subnets = get_subnets_for_room(interface.host.room) current_ips = [ip.address for ip in interface.ips] @@ -277,13 +281,10 @@ def default_response() -> ResponseReturnValue: return redirect(url_for('user.user_show', user_id=interface.host.owner_id, _anchor='hosts')) -@bp.route('//interface/create', methods=['GET', 'POST']) -@access.require('hosts_change') -def interface_create(host_id) -> ResponseReturnValue: - if (host := Host.get(host_id)) is None: - flash("Host existiert nicht.", 'error') - abort(404) - +@bp.route("//interface/create", methods=["GET", "POST"]) +@access.require("hosts_change") +def interface_create(host_id: int) -> ResponseReturnValue: + host = get_host_or_404(host_id) subnets = get_subnets_for_room(host.room) form = InterfaceForm() unused_ips = [ip for ips in get_unused_ips(subnets).values() for ip in ips] @@ -326,7 +327,7 @@ def default_response() -> ResponseReturnValue: )) -def _host_row(host, user_id) -> HostRow: +def _host_row(host: Host, user_id: int) -> HostRow: if host.room: patch_ports = host.room.connected_patch_ports switches = ", ".join( @@ -361,7 +362,7 @@ def _host_row(host, user_id) -> HostRow: @bp.route("/") -def user_hosts_json(user_id) -> ResponseReturnValue: +def user_hosts_json(user_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) return TableResponse[HostRow]( items=[_host_row(host, user_id) for host in user.hosts] @@ -369,7 +370,7 @@ def user_hosts_json(user_id) -> ResponseReturnValue: @bp.route("/interface-manufacturer/") -def interface_manufacturer_json(mac) -> ResponseReturnValue: +def interface_manufacturer_json(mac: str) -> ResponseReturnValue: if not re.match(mac_regex, mac): return abort(400) return {"manufacturer": get_interface_manufacturer(mac)} diff --git a/web/blueprints/host/tables.py b/web/blueprints/host/tables.py index 5e886f6dc..314522480 100644 --- a/web/blueprints/host/tables.py +++ b/web/blueprints/host/tables.py @@ -62,11 +62,11 @@ class InterfaceTable(BootstrapTable): ips = Column("IPs") actions = MultiBtnColumn("Aktionen", hide_if=no_hosts_change) - def __init__(self, *a, host_id=None, **kw): - table_args = kw.pop('table_args', {}) - table_args.setdefault('data-hide-pagination-info', "true") - table_args.setdefault('data-search', "false") - kw['table_args'] = table_args + def __init__(self, *a, host_id: int | None = None, **kw) -> None: + table_args = kw.pop("table_args", {}) + table_args.setdefault("data-hide-pagination-info", "true") + table_args.setdefault("data-search", "false") + kw["table_args"] = table_args super().__init__(*a, **kw) self.host_id = host_id From 7d4d33ccc725ab3552213548c18baf2b18c0aef0 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Mon, 21 Aug 2023 12:20:22 +0200 Subject: [PATCH 02/18] [web] use early return in refill_room_data --- web/blueprints/helpers/form.py | 38 +++++++++++++++++----------------- 1 file changed, 19 insertions(+), 19 deletions(-) diff --git a/web/blueprints/helpers/form.py b/web/blueprints/helpers/form.py index 26f788426..66cb9ba67 100644 --- a/web/blueprints/helpers/form.py +++ b/web/blueprints/helpers/form.py @@ -16,25 +16,25 @@ def iter_prefixed_field_names(cls: type[Form], prefix: str) -> typing.Iterator[s if hasattr(f, '_formfield') and f.startswith(prefix)) -def refill_room_data(form: SelectRoomForm, room: Room) -> None: - if room: - form.building.data = room.building - - levels = Room.q.filter_by(building_id=room.building.id).order_by(Room.level)\ - .distinct() - - form.level.choices = [(entry.level, str(entry.level)) for entry in - levels] - form.level.data = room.level - - rooms = Room.q.filter_by( - building_id=room.building.id, - level=room.level - ).order_by(Room.number).distinct() - - form.room_number.choices = [(entry.number, str(entry.number)) - for entry in rooms] - form.room_number.data = room.number +def refill_room_data(form: SelectRoomForm, room: Room | None) -> None: + if not room: + return + + form.building.data = room.building + + levels = ( + Room.q.filter_by(building_id=room.building.id).order_by(Room.level).distinct() + ) + form.level.choices = [(entry.level, str(entry.level)) for entry in levels] + form.level.data = room.level + + rooms = ( + Room.q.filter_by(building_id=room.building.id, level=room.level) + .order_by(Room.number) + .distinct() + ) + form.room_number.choices = [(entry.number, str(entry.number)) for entry in rooms] + form.room_number.data = room.number def confirmable_div(confirm_field_id: str | None, prefix: str = 'form-group-') -> str: From c50429391cae735dc42d127ad0e94b7b2c74bd3a Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Mon, 21 Aug 2023 12:48:04 +0200 Subject: [PATCH 03/18] [typing] stricter typing rules for web.blueprints.user --- pyproject.toml | 1 + web/blueprints/finance/tables.py | 4 +- web/blueprints/helpers/log_tables.py | 7 +- web/blueprints/host/tables.py | 2 +- web/blueprints/user/__init__.py | 95 ++++++++++++++++------------ web/blueprints/user/log.py | 4 +- web/blueprints/user/tables.py | 2 +- 7 files changed, 67 insertions(+), 48 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 09c1f4e5e..ae84bed80 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,6 +53,7 @@ module = [ "web.blueprints.access", "web.blueprints.facilities", "web.blueprints.host", + "web.blueprints.user", "web.blueprints.navigation", ] disallow_untyped_defs = true diff --git a/web/blueprints/finance/tables.py b/web/blueprints/finance/tables.py index c3594d9c6..c8cfe9d19 100644 --- a/web/blueprints/finance/tables.py +++ b/web/blueprints/finance/tables.py @@ -45,7 +45,7 @@ class Meta: 'data-page-list': '[5, 10, 25, 50, 100]' } - def __init__(self, *a, saldo=None, user_id=None, inverted=False, **kw): + def __init__(self, *a, saldo=None, user_id=None, inverted=False, **kw) -> None: """Init :param int user_id: An optional user_id. If set, this causes @@ -124,7 +124,7 @@ class Meta: splits = (('soll', "Soll"), ('haben', "Haben")) - def __init__(self, *a, **kw): + def __init__(self, *a, **kw) -> None: super().__init__(*a, **kw) self.table_footer_offset = 7 diff --git a/web/blueprints/helpers/log_tables.py b/web/blueprints/helpers/log_tables.py index c0dcc9c08..7eb3a4d66 100644 --- a/web/blueprints/helpers/log_tables.py +++ b/web/blueprints/helpers/log_tables.py @@ -23,7 +23,7 @@ class RefreshableTableMixin: ``{'data-show-refresh': "true"}`` is established. """ - def __init__(self, *a, **kw): + def __init__(self, *a, **kw) -> None: table_args = kw.pop("table_args", {}) table_args.setdefault("data-show-refresh", "true") kw["table_args"] = table_args @@ -48,6 +48,11 @@ class LogTableSpecific(RefreshableTableMixin, BootstrapTable): LogType = t.Literal["user", "room", "hades", "task", "all"] +LOG_TYPES = frozenset(t.get_args(LogType)) + + +def is_log_type(type: str) -> t.TypeGuard[LogType]: + return type in LOG_TYPES class LogTableRow(BaseModel): diff --git a/web/blueprints/host/tables.py b/web/blueprints/host/tables.py index 314522480..0252c917f 100644 --- a/web/blueprints/host/tables.py +++ b/web/blueprints/host/tables.py @@ -22,7 +22,7 @@ class HostTable(BootstrapTable): interface_create_link = Column("", hide_if=lambda: True) id = Column("", hide_if=lambda: True) - def __init__(self, *a, user_id=None, **kw): + def __init__(self, *a, user_id: int | None = None, **kw) -> None: table_args = kw.pop('table_args', {}) table_args.setdefault('data-load-subtables', "true") table_args.setdefault('data-detail-view', "true") diff --git a/web/blueprints/user/__init__.py b/web/blueprints/user/__init__.py index 47550c912..21624e408 100644 --- a/web/blueprints/user/__init__.py +++ b/web/blueprints/user/__init__.py @@ -117,6 +117,7 @@ LogTableExtended, LogTableSpecific, LogType, + is_log_type, LogTableRow, ) from ..finance.tables import FinanceTable, FinanceTableSplitted @@ -156,7 +157,7 @@ def overview() -> ResponseReturnValue: )) -def make_pdf_response(pdf_data, filename, inline=True) -> Response: +def make_pdf_response(pdf_data: bytes, filename: str, inline: bool = True) -> Response: """Turn pdf data into a response with appropriate headers. Content-Type: application/pdf @@ -188,15 +189,12 @@ def user_sheet() -> ResponseReturnValue: @bp.route('//datasheet') -def static_datasheet(user_id) -> ResponseReturnValue: +def static_datasheet(user_id: int) -> ResponseReturnValue: """Deliver an on-the-fly datasheet without the password. Useful for testing the layout itself. """ - user = User.get(user_id) - if user is None: - abort(404) - + user = get_user_or_404(user_id) return make_pdf_response(generate_user_sheet(True, False, user, plain_user_password="********", generation_purpose='reprint'), filename=f'user_sheet_plain_{user_id}.pdf') @@ -274,7 +272,13 @@ def json_search() -> ResponseReturnValue: ).model_dump() -def infoflags(user): +class InfoflagDict(t.TypedDict): + title: str + icon: str + val: bool + + +def infoflags(user: User) -> list[InfoflagDict]: user_status = lib.user.status(user) return [ {'title': "Mitglied", 'icon': "user", 'val': user_status.member}, @@ -287,7 +291,7 @@ def infoflags(user): @bp.route('//', methods=['GET', 'POST']) -def user_show(user_id) -> ResponseReturnValue: +def user_show(user_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) room = user.room form = UserLogEntry() @@ -425,7 +429,7 @@ def user_show(user_id) -> ResponseReturnValue: @bp.route("//account") -def user_account(user_id) -> ResponseReturnValue: +def user_account(user_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) return redirect(url_for("finance.accounts_show", account_id=user.account_id)) @@ -444,9 +448,12 @@ def _iter_user_logs(user: User, logtype: LogType) -> t.Iterator[LogTableRow]: @bp.route("//logs") @bp.route("//logs/") -def user_show_logs_json(user_id, logtype="all") -> ResponseReturnValue: +def user_show_logs_json(user_id: int, logtype: str = "all") -> ResponseReturnValue: user = get_user_or_404(user_id) + if not is_log_type(logtype): + flash(f"{logtype!r} ist kein valider logtyp", "error") + abort(404) logs = _iter_user_logs(user, logtype) def sort_key(l: LogTableRow) -> int | None: @@ -459,7 +466,9 @@ def sort_key(l: LogTableRow) -> int | None: @bp.route("//groups") @bp.route("//groups/") -def user_show_groups_json(user_id, group_filter="all") -> ResponseReturnValue: +def user_show_groups_json( + user_id: int, group_filter: str = "all" +) -> ResponseReturnValue: active_groups_only = group_filter == "active" memberships = t.cast( t.Iterable[tuple[Membership, list[str], list[str]]], @@ -517,7 +526,7 @@ def user_show_groups_json(user_id, group_filter="all") -> ResponseReturnValue: @bp.route('//add_membership', methods=['GET', 'Post']) @access.require('groups_change_membership') -def add_membership(user_id) -> ResponseReturnValue: +def add_membership(user_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) form = UserAddGroupMembership() @@ -551,15 +560,18 @@ def add_membership(user_id) -> ResponseReturnValue: user_id=user_id, form=form) +def get_membership_or_404(id: int) -> Membership: + if (membership := session.session.get(Membership, id)) is None: + flash(f"Gruppenmitgliedschaft mit ID {id} existiert nicht!", "error") + abort(404) + return membership + @bp.route('//end_membership/') @access.require('groups_change_membership') -def end_membership(user_id, membership_id) -> ResponseReturnValue: +def end_membership(user_id: int, membership_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) - membership = Membership.get(membership_id) - - if membership is None: - flash(f"Gruppenmitgliedschaft mit ID {membership_id} existiert nicht!", 'error') - abort(404) + membership = get_membership_or_404(membership_id) + assert isinstance(membership.group, PropertyGroup) if membership.user.id != user_id: flash(f"Gruppenmitgliedschaft {membership.id} gehört nicht zu Nutzer {user_id}!", 'error') @@ -582,7 +594,7 @@ def end_membership(user_id, membership_id) -> ResponseReturnValue: @bp.route('//traffic/json') @bp.route('//traffic/json/') -def json_trafficdata(user_id, days=7) -> ResponseReturnValue: +def json_trafficdata(user_id: int, days: int = 7) -> ResponseReturnValue: """Generate a JSON file to use with traffic and credit graphs. :param user_id: @@ -706,11 +718,11 @@ def default_response() -> ResponseReturnValue: @bp.route('//move', methods=['GET', 'POST']) @access.require('user_change') -def move(user_id) -> ResponseReturnValue: +def move(user_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) form = UserMoveForm() - def default_response(refill_form_data=False) -> ResponseReturnValue: + def default_response(refill_form_data: bool = False) -> ResponseReturnValue: if not form.is_submitted() or refill_form_data: if user.room is not None: refill_room_data(form, user.room) @@ -764,13 +776,8 @@ def default_response(refill_form_data=False) -> ResponseReturnValue: @bp.route('//edit_membership/', methods=['GET', 'POST']) @access.require('groups_change_membership') -def edit_membership(user_id, membership_id) -> ResponseReturnValue: - membership: Membership = Membership.get(membership_id) - - if membership is None: - flash("Gruppenmitgliedschaft mit ID {} existiert nicht!".format( - membership_id), 'error') - abort(404) +def edit_membership(user_id: int, membership_id: int) -> ResponseReturnValue: + membership = get_membership_or_404(membership_id) assert isinstance(membership.group, PropertyGroup) if membership.group.permission_level > current_user.permission_level: @@ -816,7 +823,7 @@ def edit_membership(user_id, membership_id) -> ResponseReturnValue: @bp.route('//edit', methods=['GET', 'POST']) @access.require('user_change') -def edit_user(user_id) -> ResponseReturnValue: +def edit_user(user_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) form = UserEditForm(obj=user, person_id=user.swdd_person_id) @@ -915,7 +922,7 @@ def search() -> ResponseReturnValue: @bp.route('//reset_password', methods=['GET', 'POST']) @access.require('user_change') -def reset_password(user_id) -> ResponseReturnValue: +def reset_password(user_id: int) -> ResponseReturnValue: form = UserResetPasswordForm() my_user = get_user_or_404(user_id) @@ -945,14 +952,18 @@ def reset_password(user_id) -> ResponseReturnValue: @bp.route('//reset_wifi_password', methods=['GET', 'POST']) @access.require('user_change') -def reset_wifi_password(user_id) -> ResponseReturnValue: +def reset_wifi_password(user_id: int) -> ResponseReturnValue: + user = get_user_or_404(user_id) form = UserResetPasswordForm() - myUser = User.get(user_id) if form.validate_on_submit(): - plain_password = lib.user.reset_wifi_password(myUser, processor=current_user) - sheet = lib.user.store_user_sheet(False, True, user=myUser, - plain_wifi_password=plain_password, - generation_purpose='password reset') + plain_password = lib.user.reset_wifi_password(user, processor=current_user) + sheet = lib.user.store_user_sheet( + False, + True, + user=user, + plain_wifi_password=plain_password, + generation_purpose="password reset", + ) session.session.commit() @@ -965,7 +976,7 @@ def reset_wifi_password(user_id) -> ResponseReturnValue: @bp.route('//block', methods=['GET', 'POST']) @access.require('user_change') -def block(user_id) -> ResponseReturnValue: +def block(user_id: int) -> ResponseReturnValue: form = UserSuspendForm() myUser = get_user_or_404(user_id) if form.validate_on_submit(): @@ -993,7 +1004,7 @@ def block(user_id) -> ResponseReturnValue: @bp.route('//unblock', methods=['GET', 'POST']) @access.require('user_change') -def unblock(user_id) -> ResponseReturnValue: +def unblock(user_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) try: @@ -1010,7 +1021,7 @@ def unblock(user_id) -> ResponseReturnValue: @bp.route('//move_out', methods=['GET', 'POST']) @access.require('user_change') -def move_out(user_id) -> ResponseReturnValue: +def move_out(user_id: int) -> ResponseReturnValue: form = UserMoveOutForm() user = get_user_or_404(user_id) @@ -1051,7 +1062,7 @@ def default_response() -> ResponseReturnValue: @bp.route('//move_in', methods=['GET', 'POST']) @access.require('user_change') -def move_in(user_id) -> ResponseReturnValue: +def move_in(user_id: int) -> ResponseReturnValue: form = UserMoveInForm() user = get_user_or_404(user_id) @@ -1096,7 +1107,7 @@ def default_response() -> ResponseReturnValue: @bp.route('/json/room-history') -def room_history_json(user_id) -> ResponseReturnValue: +def room_history_json(user_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) return TableResponse[RoomHistoryRow]( items=[ @@ -1118,7 +1129,7 @@ def room_history_json(user_id) -> ResponseReturnValue: @bp.route('/json/tenancies') -def tenancies_json(user_id) -> ResponseReturnValue: +def tenancies_json(user_id: int) -> ResponseReturnValue: user = get_user_or_404(user_id) return TableResponse[TenancyRow]( items=[ diff --git a/web/blueprints/user/log.py b/web/blueprints/user/log.py index 618189f73..9697b9056 100644 --- a/web/blueprints/user/log.py +++ b/web/blueprints/user/log.py @@ -5,6 +5,7 @@ a user. """ import logging +import typing as t from sqlalchemy.orm import Query @@ -16,6 +17,7 @@ from pycroft.model.user import User from ..helpers.log import format_hades_log_entry, format_hades_disabled, \ format_user_not_connected, format_hades_error, format_hades_timeout +from ..helpers.log_tables import LogTableRow logger = logging.getLogger(__name__) @@ -75,7 +77,7 @@ def is_user_connected(user): return True -def formatted_user_hades_logs(user): +def formatted_user_hades_logs(user: User) -> t.Iterator[LogTableRow]: """Iterate over the user's hades logs if configured correctly In case of misconfiguration, i.e. if diff --git a/web/blueprints/user/tables.py b/web/blueprints/user/tables.py index 9f3a24aa4..b057682d4 100644 --- a/web/blueprints/user/tables.py +++ b/web/blueprints/user/tables.py @@ -30,7 +30,7 @@ class MembershipTable(BootstrapTable): ends_at = DateColumn("Ende") actions = MultiBtnColumn("Aktionen", hide_if=no_membership_change) - def __init__(self, *a, user_id=None, **kw): + def __init__(self, *a, user_id: int | None = None, **kw) -> None: super().__init__(*a, **kw) self.user_id = user_id From 092088172e2037ec8cbd4869b810aab292f6017e Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Wed, 23 Aug 2023 10:12:19 +0200 Subject: [PATCH 04/18] [typing] stricter rules for web.blueprints.infrastructure --- pyproject.toml | 1 + web/blueprints/infrastructure/__init__.py | 49 +++++++++++++---------- web/blueprints/infrastructure/tables.py | 2 +- 3 files changed, 29 insertions(+), 23 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index ae84bed80..08486fb21 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -53,6 +53,7 @@ module = [ "web.blueprints.access", "web.blueprints.facilities", "web.blueprints.host", + "web.blueprints.infrastructure", "web.blueprints.user", "web.blueprints.navigation", ] diff --git a/web/blueprints/infrastructure/__init__.py b/web/blueprints/infrastructure/__init__.py index 1a62a6632..c86766ee9 100644 --- a/web/blueprints/infrastructure/__init__.py +++ b/web/blueprints/infrastructure/__init__.py @@ -14,7 +14,7 @@ from flask.typing import ResponseValue from flask_login import current_user from flask_wtf import FlaskForm as Form -from ipaddr import IPAddress +from ipaddr import IPAddress, _BaseIP from sqlalchemy.orm import joinedload from pycroft.lib.infrastructure import create_switch, \ @@ -26,7 +26,7 @@ from pycroft.model import session from pycroft.model.facilities import Room from pycroft.model.host import Switch, SwitchPort -from pycroft.model.net import VLAN +from pycroft.model.net import VLAN, Subnet from pycroft.model.port import PatchPort from web.blueprints.access import BlueprintAccess from web.blueprints.infrastructure.forms import SwitchForm, SwitchPortForm @@ -56,7 +56,7 @@ def subnets() -> ResponseValue: subnet_table=SubnetTable(data_url=url_for(".subnets_json"))) -def format_address_range(base_address, amount): +def format_address_range(base_address: _BaseIP, amount: int) -> str: if amount == 0: raise ValueError if abs(amount) == 1: @@ -66,7 +66,7 @@ def format_address_range(base_address, amount): return f'{str(base_address + amount + 1)} - {str(base_address)}' -def format_reserved_addresses(subnet): +def format_reserved_addresses(subnet: Subnet) -> list[str]: reserved = [] if subnet.reserved_addresses_bottom: reserved.append( @@ -135,8 +135,8 @@ def switches_json() -> ResponseValue: @bp.route('/switch/show/') -def switch_show(switch_id) -> ResponseValue: - switch = Switch.get(switch_id) +def switch_show(switch_id: int) -> ResponseValue: + switch = session.session.get(Switch, switch_id) if not switch: flash(f"Switch mit ID {switch_id} nicht gefunden!", "error") return redirect(url_for('.switches')) @@ -150,14 +150,19 @@ def switch_show(switch_id) -> ResponseValue: @bp.route('/switch/show//json') -def switch_show_json(switch_id) -> ResponseValue: - switch = Switch.q.options( - joinedload(Switch.ports).joinedload(SwitchPort.patch_port).joinedload( - PatchPort.room)).get(switch_id) +def switch_show_json(switch_id: int) -> ResponseValue: + switch = session.session.get( + Switch, + switch_id, + options=[ + joinedload(Switch.ports) + .joinedload(SwitchPort.patch_port) + .joinedload(PatchPort.room) + ], + ) if not switch: abort(404) - switch_port_list = switch.ports - switch_port_list = sort_ports(switch_port_list) + switch_port_list = sort_ports(switch.ports) return TableResponse[PortRow]( items=[ @@ -229,7 +234,7 @@ def switch_create() -> ResponseValue: @bp.route('/switch//edit', methods=['GET', 'POST']) @access.require('infrastructure_change') -def switch_edit(switch_id) -> ResponseValue: +def switch_edit(switch_id: int) -> ResponseValue: sess = session.session if not (switch := sess.get(Switch, switch_id)): flash(f"Switch mit ID {switch_id} nicht gefunden!", "error") @@ -260,7 +265,7 @@ def switch_edit(switch_id) -> ResponseValue: @bp.route('/switch//delete', methods=['GET', 'POST']) @access.require('infrastructure_change') -def switch_delete(switch_id) -> ResponseValue: +def switch_delete(switch_id: int) -> ResponseValue: sess = session.session if not (switch := sess.get(Switch, switch_id)): flash(f"Switch mit ID {switch_id} nicht gefunden!", "error") @@ -289,8 +294,8 @@ def switch_delete(switch_id) -> ResponseValue: @bp.route('/switch//port/create', methods=['GET', 'POST']) @access.require('infrastructure_change') -def switch_port_create(switch_id) -> ResponseValue: - switch = Switch.get(switch_id) +def switch_port_create(switch_id: int) -> ResponseValue: + switch = session.session.get(Switch, switch_id) if not switch: flash(f"Switch mit ID {switch_id} nicht gefunden!", "error") @@ -338,9 +343,9 @@ def switch_port_create(switch_id) -> ResponseValue: @bp.route('/switch//port//edit', methods=['GET', 'POST']) @access.require('infrastructure_change') -def switch_port_edit(switch_id, switch_port_id) -> ResponseValue: - switch = Switch.get(switch_id) - switch_port = SwitchPort.get(switch_port_id) +def switch_port_edit(switch_id: int, switch_port_id: int) -> ResponseValue: + switch = session.session.get(Switch, switch_id) + switch_port = session.session.get(SwitchPort, switch_port_id) if not switch: flash(f"Switch mit ID {switch_id} nicht gefunden!", "error") @@ -398,9 +403,9 @@ def switch_port_edit(switch_id, switch_port_id) -> ResponseValue: @bp.route('/switch//port//delete', methods=['GET', 'POST']) @access.require('infrastructure_change') -def switch_port_delete(switch_id, switch_port_id) -> ResponseValue: - switch = Switch.get(switch_id) - switch_port = SwitchPort.get(switch_port_id) +def switch_port_delete(switch_id: int, switch_port_id: int) -> ResponseValue: + switch = session.session.get(Switch, switch_id) + switch_port = session.session.get(SwitchPort, switch_port_id) if not switch: flash(f"Switch mit ID {switch_id} nicht gefunden!", "error") diff --git a/web/blueprints/infrastructure/tables.py b/web/blueprints/infrastructure/tables.py index 075ac6065..2d81fb84b 100644 --- a/web/blueprints/infrastructure/tables.py +++ b/web/blueprints/infrastructure/tables.py @@ -82,7 +82,7 @@ class Meta: 'data-sort-name': 'switchport_name', } - def __init__(self, *a, switch_id=None, **kw): + def __init__(self, *a, switch_id=None, **kw) -> None: super().__init__(*a, **kw) self.switch_id = switch_id From a20b27197bb38d9cd8c6157aa4136b67ee72093e Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Wed, 23 Aug 2023 10:28:23 +0200 Subject: [PATCH 05/18] [typing] replace remaining `ModelBase.get()` usages --- web/blueprints/finance/__init__.py | 34 +++++++++++++-------------- web/blueprints/login/__init__.py | 3 ++- web/blueprints/properties/__init__.py | 10 ++++---- web/blueprints/task/__init__.py | 2 +- 4 files changed, 25 insertions(+), 24 deletions(-) diff --git a/web/blueprints/finance/__init__.py b/web/blueprints/finance/__init__.py index b5e4a1920..a40252cbd 100644 --- a/web/blueprints/finance/__init__.py +++ b/web/blueprints/finance/__init__.py @@ -275,7 +275,7 @@ def display_form_response( if not form.validate(): return display_form_response(imported) - bank_account = BankAccount.get(form.account.data) + bank_account = session.get(BankAccount, form.account.data) # set start_date, end_date if form.start_date.data is None: @@ -378,7 +378,7 @@ def bank_accounts_import_errors() -> ResponseReturnValue: @bp.route('/bank-accounts/importerrors/', methods=['GET', 'POST']) @access.require('finance_change') def fix_import_error(error_id) -> ResponseReturnValue: - error = MT940Error.get(error_id) + error = session.get(MT940Error, error_id) form = FixMT940Form() imported = ImportedTransactions([], [], []) new_exception = None @@ -455,7 +455,7 @@ def bank_accounts_create() -> ResponseReturnValue: @bp.route('/bank-account-activities/', methods=["GET", "POST"]) def bank_account_activities_edit(activity_id) -> ResponseReturnValue: - activity = BankAccountActivity.get(activity_id) + activity = session.get(BankAccountActivity, activity_id) if activity is None: flash(f"Bankbewegung mit ID {activity_id} existiert nicht!", 'error') @@ -643,7 +643,7 @@ def accounts_list() -> ResponseReturnValue: @bp.route('/account//toggle-legacy') @access.require('finance_change') def account_toggle_legacy(account_id) -> ResponseReturnValue: - account = Account.get(account_id) + account = session.get(Account, account_id) if not account: abort(404) @@ -768,7 +768,7 @@ def accounts_show_json(account_id) -> ResponseReturnValue: if sort_by.startswith("soll_") or sort_order.startswith("haben_"): sort_by = '_'.join(sort_by.split('_')[1:]) - account = Account.get(account_id) or abort(404) + account = session.get(Account, account_id) or abort(404) total = Split.q.join(Transaction).filter(Split.account == account).count() @@ -803,7 +803,7 @@ def rows_from_query(query): @bp.route('/transactions/') def transactions_show(transaction_id) -> ResponseReturnValue: - transaction = Transaction.get(transaction_id) + transaction = session.get(Transaction, transaction_id) if transaction is None: abort(404) @@ -821,7 +821,7 @@ def transactions_show(transaction_id) -> ResponseReturnValue: @bp.route('/transactions//json') def transactions_show_json(transaction_id) -> ResponseReturnValue: - transaction = Transaction.get(transaction_id) + transaction = session.get(Transaction, transaction_id) return TransactionSplitResponse( description=transaction.description, items=[ @@ -943,7 +943,7 @@ def transactions_unconfirmed_json() -> ResponseReturnValue: @bp.route('/transaction//confirm', methods=['GET', 'POST']) @access.require('finance_change') def transaction_confirm(transaction_id) -> ResponseReturnValue: - transaction = Transaction.get(transaction_id) + transaction = session.get(Transaction, transaction_id) if transaction is None: flash("Transaktion existiert nicht.", 'error') @@ -977,7 +977,7 @@ def transactions_confirm_selected() -> ResponseReturnValue: for id in ids: if isinstance(id, int): - transaction = Transaction.get(int(id)) + transaction = session.get(Transaction, int(id)) if transaction: lib.finance.transaction_confirm(transaction, current_user) else: @@ -1019,7 +1019,7 @@ def default_response() -> str: @bp.route('/transaction//delete', methods=['GET', 'POST']) @access.require('finance_change') def transaction_delete(transaction_id) -> ResponseReturnValue: - transaction = Transaction.get(transaction_id) + transaction = session.get(Transaction, transaction_id) if transaction is None: flash("Transaktion existiert nicht.", 'error') @@ -1116,10 +1116,10 @@ def transactions_all_json() -> ResponseReturnValue: def transactions_create() -> ResponseReturnValue: form = TransactionCreateForm() if form.validate_on_submit(): - splits = [( - Account.get(split_form.account_id.data), - split_form.amount.data - ) for split_form in form.splits] + splits = [ + (session.get(Account, split_form.account_id.data), split_form.amount.data) + for split_form in form.splits + ] transaction = finance.complex_transaction( description=form.description.data, author=current_user, @@ -1158,7 +1158,7 @@ def accounts_create() -> ResponseReturnValue: @bp.route("/membership_fee//book", methods=['GET', 'POST']) @access.require('finance_change') def membership_fee_book(fee_id) -> ResponseReturnValue: - fee = MembershipFee.get(fee_id) + fee = session.get(MembershipFee, fee_id) if fee is None: flash('Ein Beitrag mit dieser ID existiert nicht!', 'error') @@ -1182,7 +1182,7 @@ def membership_fee_book(fee_id) -> ResponseReturnValue: @bp.route("/membership_fee//users_due_json") def membership_fee_users_due_json(fee_id) -> ResponseReturnValue: - fee = MembershipFee.get(fee_id) + fee = session.get(MembershipFee, fee_id) if fee is None: abort(404) @@ -1316,7 +1316,7 @@ def membership_fee_create() -> ResponseReturnValue: @bp.route('/membership_fee//edit', methods=("GET", "POST")) @access.require('finance_change') def membership_fee_edit(fee_id) -> ResponseReturnValue: - fee = MembershipFee.get(fee_id) + fee = session.get(MembershipFee, fee_id) if fee is None: flash('Ein Beitrag mit dieser ID existiert nicht!', 'error') diff --git a/web/blueprints/login/__init__.py b/web/blueprints/login/__init__.py index c81d909a9..326d91c06 100644 --- a/web/blueprints/login/__init__.py +++ b/web/blueprints/login/__init__.py @@ -18,6 +18,7 @@ AnonymousUserMixin, LoginManager, current_user, login_required, login_user, logout_user) +from pycroft.model.session import session from pycroft.model.user import User from web.blueprints.login.forms import LoginForm @@ -37,7 +38,7 @@ class AnonymousUser(AnonymousUserMixin): @login_manager.user_loader def load_user(userid): - return User.get(userid) + return session.get(User, userid) @bp.route("/login", methods=("GET", "POST")) diff --git a/web/blueprints/properties/__init__.py b/web/blueprints/properties/__init__.py index 05c0956de..9a59de86c 100644 --- a/web/blueprints/properties/__init__.py +++ b/web/blueprints/properties/__init__.py @@ -74,7 +74,7 @@ def property_group_create() -> ResponseValue: @bp.route('/property_group//grant/') @access.require('groups_change') def property_group_grant_property(group_id, property_name) -> ResponseValue: - property_group = PropertyGroup.get(group_id) + property_group = session.session.get(PropertyGroup, group_id) if property_group is None: flash(f"Eigenschaftengruppe mit ID {group_id} existiert nicht!", 'error') @@ -90,7 +90,7 @@ def property_group_grant_property(group_id, property_name) -> ResponseValue: @bp.route('/property_group//deny/') @access.require('groups_change') def property_group_deny_property(group_id, property_name) -> ResponseValue: - property_group = PropertyGroup.get(group_id) + property_group = session.session.get(PropertyGroup, group_id) if property_group is None: flash(f"Eigenschaftengruppe mit ID {group_id} existiert nicht!", 'error') @@ -106,7 +106,7 @@ def property_group_deny_property(group_id, property_name) -> ResponseValue: @bp.route('/property_group//remove/') @access.require('groups_change') def property_group_remove_property(group_id, property_name) -> ResponseValue: - group = PropertyGroup.get(group_id) + group = session.session.get(PropertyGroup, group_id) if group is None: flash(f"Eigenschaftengruppe mit ID {group_id} existiert nicht!", 'error') @@ -122,7 +122,7 @@ def property_group_remove_property(group_id, property_name) -> ResponseValue: @bp.route('/property_group//edit', methods=['GET', 'POST']) @access.require('groups_change') def property_group_edit(group_id) -> ResponseValue: - group = PropertyGroup.get(group_id) + group = session.session.get(PropertyGroup, group_id) if group is None: flash(f"Eigenschaftengruppe mit ID {group_id} existiert nicht!", 'error') @@ -154,7 +154,7 @@ def property_group_edit(group_id) -> ResponseValue: @bp.route('/property_group//delete', methods=['GET', 'POST']) @access.require('groups_change') def property_group_delete(group_id) -> ResponseValue: - group = PropertyGroup.get(group_id) + group = session.session.get(PropertyGroup, group_id) if group is None: flash(f"Eigenschaftengruppe mit ID {group_id} existiert nicht!", 'error') diff --git a/web/blueprints/task/__init__.py b/web/blueprints/task/__init__.py index 59440d403..7010011df 100644 --- a/web/blueprints/task/__init__.py +++ b/web/blueprints/task/__init__.py @@ -37,7 +37,7 @@ def format_parameters(parameters): # Replace building_id by the buildings short name if bid := parameters.get("building_id"): - if building := Building.get(bid): + if building := session.session.get(Building, bid): parameters["building"] = building.short_name del parameters["building_id"] return parameters From dd7353d223806a40dda8614cdaefe6d41aabec83 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:13:04 +0200 Subject: [PATCH 06/18] [typing] enable stricter mypy rules for whole `blueprints` package Refs #559 --- pyproject.toml | 7 +------ 1 file changed, 1 insertion(+), 6 deletions(-) diff --git a/pyproject.toml b/pyproject.toml index 08486fb21..ff66b1b12 100644 --- a/pyproject.toml +++ b/pyproject.toml @@ -50,12 +50,7 @@ module = [ "pycroft.helpers.user", "pycroft.helpers.utc", "web.blueprints", - "web.blueprints.access", - "web.blueprints.facilities", - "web.blueprints.host", - "web.blueprints.infrastructure", - "web.blueprints.user", - "web.blueprints.navigation", + "web.blueprints.*", ] disallow_untyped_defs = true disallow_untyped_calls = true From 65978a529d3bae80687b0658cdcbff37d8200883 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:14:38 +0200 Subject: [PATCH 07/18] [typing] add type hints to `bp.login` --- stubs/flask_login/__about__.pyi | 7 ++++++ stubs/flask_login/__init__.pyi | 5 ++++ stubs/flask_login/config.pyi | 16 +++++++++++++ stubs/flask_login/login_manager.pyi | 36 +++++++++++++++++++++++++++++ stubs/flask_login/mixins.pyi | 22 ++++++++++++++++++ stubs/flask_login/signals.pyi | 13 +++++++++++ stubs/flask_login/utils.pyi | 23 ++++++++++++++++++ web/blueprints/login/__init__.py | 2 +- 8 files changed, 123 insertions(+), 1 deletion(-) create mode 100644 stubs/flask_login/__about__.pyi create mode 100644 stubs/flask_login/__init__.pyi create mode 100644 stubs/flask_login/config.pyi create mode 100644 stubs/flask_login/login_manager.pyi create mode 100644 stubs/flask_login/mixins.pyi create mode 100644 stubs/flask_login/signals.pyi create mode 100644 stubs/flask_login/utils.pyi diff --git a/stubs/flask_login/__about__.pyi b/stubs/flask_login/__about__.pyi new file mode 100644 index 000000000..f3270d2bb --- /dev/null +++ b/stubs/flask_login/__about__.pyi @@ -0,0 +1,7 @@ +from _typeshed import Incomplete + +__description__: str +__url__: str +__version_info__: Incomplete +__author_email__: str +__maintainer__: str diff --git a/stubs/flask_login/__init__.pyi b/stubs/flask_login/__init__.pyi new file mode 100644 index 000000000..c25452918 --- /dev/null +++ b/stubs/flask_login/__init__.pyi @@ -0,0 +1,5 @@ +from .config import AUTH_HEADER_NAME as AUTH_HEADER_NAME, COOKIE_DURATION as COOKIE_DURATION, COOKIE_HTTPONLY as COOKIE_HTTPONLY, COOKIE_NAME as COOKIE_NAME, COOKIE_SECURE as COOKIE_SECURE, ID_ATTRIBUTE as ID_ATTRIBUTE, LOGIN_MESSAGE as LOGIN_MESSAGE, LOGIN_MESSAGE_CATEGORY as LOGIN_MESSAGE_CATEGORY, REFRESH_MESSAGE as REFRESH_MESSAGE, REFRESH_MESSAGE_CATEGORY as REFRESH_MESSAGE_CATEGORY +from .login_manager import LoginManager as LoginManager +from .mixins import AnonymousUserMixin as AnonymousUserMixin, UserMixin as UserMixin +from .signals import session_protected as session_protected, user_accessed as user_accessed, user_loaded_from_cookie as user_loaded_from_cookie, user_loaded_from_request as user_loaded_from_request, user_logged_in as user_logged_in, user_logged_out as user_logged_out, user_login_confirmed as user_login_confirmed, user_needs_refresh as user_needs_refresh, user_unauthorized as user_unauthorized +from .utils import confirm_login as confirm_login, current_user as current_user, decode_cookie as decode_cookie, encode_cookie as encode_cookie, fresh_login_required as fresh_login_required, login_fresh as login_fresh, login_remembered as login_remembered, login_required as login_required, login_url as login_url, login_user as login_user, logout_user as logout_user, make_next_param as make_next_param, set_login_view as set_login_view diff --git a/stubs/flask_login/config.pyi b/stubs/flask_login/config.pyi new file mode 100644 index 000000000..62408bf66 --- /dev/null +++ b/stubs/flask_login/config.pyi @@ -0,0 +1,16 @@ +from _typeshed import Incomplete + +COOKIE_NAME: str +COOKIE_DURATION: Incomplete +COOKIE_SECURE: bool +COOKIE_HTTPONLY: bool +COOKIE_SAMESITE: Incomplete +LOGIN_MESSAGE: str +LOGIN_MESSAGE_CATEGORY: str +REFRESH_MESSAGE: str +REFRESH_MESSAGE_CATEGORY: str +ID_ATTRIBUTE: str +AUTH_HEADER_NAME: str +SESSION_KEYS: Incomplete +EXEMPT_METHODS: Incomplete +USE_SESSION_FOR_NEXT: bool diff --git a/stubs/flask_login/login_manager.pyi b/stubs/flask_login/login_manager.pyi new file mode 100644 index 000000000..5bb679c93 --- /dev/null +++ b/stubs/flask_login/login_manager.pyi @@ -0,0 +1,36 @@ +import typing as t +from .config import AUTH_HEADER_NAME as AUTH_HEADER_NAME, COOKIE_DURATION as COOKIE_DURATION, COOKIE_HTTPONLY as COOKIE_HTTPONLY, COOKIE_NAME as COOKIE_NAME, COOKIE_SAMESITE as COOKIE_SAMESITE, COOKIE_SECURE as COOKIE_SECURE, ID_ATTRIBUTE as ID_ATTRIBUTE, LOGIN_MESSAGE as LOGIN_MESSAGE, LOGIN_MESSAGE_CATEGORY as LOGIN_MESSAGE_CATEGORY, REFRESH_MESSAGE as REFRESH_MESSAGE, REFRESH_MESSAGE_CATEGORY as REFRESH_MESSAGE_CATEGORY, SESSION_KEYS as SESSION_KEYS, USE_SESSION_FOR_NEXT as USE_SESSION_FOR_NEXT +from .mixins import AnonymousUserMixin as AnonymousUserMixin +from .signals import session_protected as session_protected, user_accessed as user_accessed, user_loaded_from_cookie as user_loaded_from_cookie, user_loaded_from_request as user_loaded_from_request, user_needs_refresh as user_needs_refresh, user_unauthorized as user_unauthorized +from .utils import decode_cookie as decode_cookie, encode_cookie as encode_cookie, expand_login_view as expand_login_view, make_next_param as make_next_param +from _typeshed import Incomplete + +class LoginManager: + anonymous_user: Incomplete + login_view: Incomplete + blueprint_login_views: Incomplete + login_message: Incomplete + login_message_category: Incomplete + refresh_view: Incomplete + needs_refresh_message: Incomplete + needs_refresh_message_category: Incomplete + session_protection: str + localize_callback: Incomplete + unauthorized_callback: Incomplete + needs_refresh_callback: Incomplete + id_attribute: Incomplete + def __init__(self, app: Incomplete | None = ..., add_context_processor: bool = ...) -> None: ... + def setup_app(self, app, add_context_processor: bool = ...) -> None: ... + def init_app(self, app, add_context_processor: bool = ...) -> None: ... + def unauthorized(self): ... + _C = t.TypeVar("_C") + def user_loader(self, callback: _C) -> _C: ... + @property + def user_callback(self): ... + def request_loader(self, callback): ... + @property + def request_callback(self): ... + def unauthorized_handler(self, callback): ... + def needs_refresh_handler(self, callback): ... + def needs_refresh(self): ... + def header_loader(self, callback): ... diff --git a/stubs/flask_login/mixins.pyi b/stubs/flask_login/mixins.pyi new file mode 100644 index 000000000..c8d331fc5 --- /dev/null +++ b/stubs/flask_login/mixins.pyi @@ -0,0 +1,22 @@ +from _typeshed import Incomplete + +class UserMixin: + __hash__: Incomplete + @property + def is_active(self): ... + @property + def is_authenticated(self): ... + @property + def is_anonymous(self): ... + def get_id(self): ... + def __eq__(self, other): ... + def __ne__(self, other): ... + +class AnonymousUserMixin: + @property + def is_authenticated(self): ... + @property + def is_active(self): ... + @property + def is_anonymous(self): ... + def get_id(self) -> None: ... diff --git a/stubs/flask_login/signals.pyi b/stubs/flask_login/signals.pyi new file mode 100644 index 000000000..75e4db280 --- /dev/null +++ b/stubs/flask_login/signals.pyi @@ -0,0 +1,13 @@ +from _typeshed import Incomplete + +user_logged_in: Incomplete +user_logged_out: Incomplete +user_loaded_from_cookie: Incomplete +user_loaded_from_request: Incomplete +user_login_confirmed: Incomplete +user_unauthorized: Incomplete +user_needs_refresh: Incomplete +user_accessed: Incomplete +session_protected: Incomplete + +def __getattr__(name): ... diff --git a/stubs/flask_login/utils.pyi b/stubs/flask_login/utils.pyi new file mode 100644 index 000000000..747e3f8a1 --- /dev/null +++ b/stubs/flask_login/utils.pyi @@ -0,0 +1,23 @@ +import typing as t +from .config import COOKIE_NAME as COOKIE_NAME, EXEMPT_METHODS as EXEMPT_METHODS +from .signals import user_logged_in as user_logged_in, user_logged_out as user_logged_out, user_login_confirmed as user_login_confirmed +from _typeshed import Incomplete + +current_user: Incomplete + +def encode_cookie(payload, key: Incomplete | None = ...): ... +def decode_cookie(cookie, key: Incomplete | None = ...): ... +def make_next_param(login_url, current_url): ... +def expand_login_view(login_view): ... +def login_url(login_view, next_url: Incomplete | None = ..., next_field: str = ...): ... +def login_fresh(): ... +def login_remembered(): ... +def login_user(user, remember: bool = ..., duration: Incomplete | None = ..., force: bool = ..., fresh: bool = ...): ... +def logout_user() -> bool: ... +def confirm_login() -> None: ... + +_F = t.TypeVar("_F") + +def login_required(func: _F) -> _F: ... +def fresh_login_required(func): ... +def set_login_view(login_view, blueprint: Incomplete | None = ...) -> None: ... diff --git a/web/blueprints/login/__init__.py b/web/blueprints/login/__init__.py index 326d91c06..84e9950c0 100644 --- a/web/blueprints/login/__init__.py +++ b/web/blueprints/login/__init__.py @@ -37,7 +37,7 @@ class AnonymousUser(AnonymousUserMixin): @login_manager.user_loader -def load_user(userid): +def load_user(userid: int) -> User | None: return session.get(User, userid) From 33b1329e4e9a3aa5b3c862b1167787e9ac2ec06e Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:17:08 +0200 Subject: [PATCH 08/18] [typing] stronger typing for `finance` blueprint --- pycroft/lib/finance/matching.py | 9 +- web/blueprints/finance/__init__.py | 161 ++++++++++++++++++++++------- web/blueprints/finance/forms.py | 14 +-- 3 files changed, 136 insertions(+), 48 deletions(-) diff --git a/pycroft/lib/finance/matching.py b/pycroft/lib/finance/matching.py index 8df7cbbe5..0a1cb1f81 100644 --- a/pycroft/lib/finance/matching.py +++ b/pycroft/lib/finance/matching.py @@ -3,6 +3,7 @@ # the Apache License, Version 2.0. See the LICENSE file for details import logging import re +import typing as t from typing import Callable, TypeVar from sqlalchemy import select @@ -17,9 +18,11 @@ T = TypeVar("T") -def match_activities() -> ( - tuple[dict[BankAccountActivity, User], dict[BankAccountActivity, Account]] -): +UserMatching: t.TypeAlias = dict[BankAccountActivity, User] +AccountMatching: t.TypeAlias = dict[BankAccountActivity, Account] + + +def match_activities() -> tuple[UserMatching, AccountMatching]: """For all unmatched transactions, determine which user or team they should be matched with.""" matching: dict[BankAccountActivity, User] = {} team_matching: dict[BankAccountActivity, Account] = {} diff --git a/web/blueprints/finance/__init__.py b/web/blueprints/finance/__init__.py index a40252cbd..c6a52b960 100644 --- a/web/blueprints/finance/__init__.py +++ b/web/blueprints/finance/__init__.py @@ -10,6 +10,7 @@ :copyright: (c) 2012 by AG DSN. """ import typing as t +from decimal import Decimal from collections.abc import Iterable from datetime import date from datetime import timedelta, datetime @@ -38,10 +39,19 @@ from flask_login import current_user from flask_wtf import FlaskForm from mt940.models import Transaction as MT940Transaction -from sqlalchemy import or_, Text, cast, ColumnClause, FromClause +from sqlalchemy import ( + or_, + Text, + cast, + ColumnClause, + FromClause, + Select, + Over, + ColumnElement, +) from sqlalchemy.orm.exc import NoResultFound, MultipleResultsFound from sqlalchemy.sql.expression import literal_column, func, select, Join -from wtforms import BooleanField +from wtforms import BooleanField, FormField, Field from pycroft import config, lib from pycroft.exc import PycroftException @@ -65,6 +75,7 @@ get_last_membership_fee, ) from pycroft.lib.finance.fints import get_fints_transactions +from pycroft.lib.finance.matching import UserMatching, AccountMatching from pycroft.lib.mail import MemberNegativeBalance from pycroft.lib.user import encode_type2_user_id, user_send_mails from pycroft.model.finance import Account, Transaction @@ -175,7 +186,7 @@ def bank_accounts_list_json() -> ResponseReturnValue: @bp.route('/bank-accounts/activities/json') def bank_accounts_activities_json() -> ResponseReturnValue: - def actions(activity_id) -> list[BtnColResponse]: + def actions(activity_id: int) -> list[BtnColResponse]: return [ BtnColResponse( href=url_for(".bank_account_activities_edit", activity_id=activity_id), @@ -230,7 +241,7 @@ def bank_accounts_errors_json() -> ResponseReturnValue: from contextlib import contextmanager @contextmanager -def flash_fints_errors(): +def flash_fints_errors() -> t.Iterator[None]: try: yield except (FinTSDialogError, FinTSClientPINError) as e: @@ -377,7 +388,7 @@ def bank_accounts_import_errors() -> ResponseReturnValue: @bp.route('/bank-accounts/importerrors/', methods=['GET', 'POST']) @access.require('finance_change') -def fix_import_error(error_id) -> ResponseReturnValue: +def fix_import_error(error_id: int) -> ResponseReturnValue: error = session.get(MT940Error, error_id) form = FixMT940Form() imported = ImportedTransactions([], [], []) @@ -454,7 +465,7 @@ def bank_accounts_create() -> ResponseReturnValue: @bp.route('/bank-account-activities/', methods=["GET", "POST"]) -def bank_account_activities_edit(activity_id) -> ResponseReturnValue: +def bank_account_activities_edit(activity_id: int) -> ResponseReturnValue: activity = session.get(BankAccountActivity, activity_id) if activity is None: @@ -537,19 +548,71 @@ def bank_account_activities_match() -> ResponseReturnValue: activities_team=matched_activities_team) -def _create_field_list_and_matched_activities_dict(matching, prefix): - matched_activities = {} +class UserMatch(t.TypedDict): + purpose: str + name: str + user: User + amount: Decimal | int + + +class TeamMatch(t.TypedDict): + purpose: str + name: str + team: Account + amount: Decimal | int + + +BooleanFieldList = list[t.Tuple[str, BooleanField]] + + +@t.overload +def _create_field_list_and_matched_activities_dict( + matching: t.Mapping[BankAccountActivity, Account], prefix: t.Literal["team"] +) -> tuple[BooleanFieldList, dict[str, TeamMatch]]: + ... + + +@t.overload +def _create_field_list_and_matched_activities_dict( + matching: t.Mapping[BankAccountActivity, User], prefix: t.Literal["user"] +) -> tuple[BooleanFieldList, dict[str, UserMatch]]: + ... + + +def _create_field_list_and_matched_activities_dict( + matching: t.Mapping[BankAccountActivity, User] + | t.Mapping[BankAccountActivity, Account], + prefix: t.Literal["user", "team"], +) -> ( + tuple[BooleanFieldList, dict[str, UserMatch]] + | tuple[BooleanFieldList, dict[str, TeamMatch]] +): + # compatibility of this implementation with the overload signatures is hard to verify; + # mypy is not too good at dependent typing + matched_activities: dict[str, UserMatch] | dict[str, TeamMatch] + if prefix == "user": + matched_activities = t.cast(dict[str, UserMatch], {}) + elif prefix == "team": + matched_activities = t.cast(dict[str, TeamMatch], {}) + else: + raise ValueError(f"Invalid Prefix: {prefix}") + field_list = [] + # mypy cannot verify that the type of `entity` is constant across iterations for activity, entity in matching.items(): matched_activities[f"{prefix}-{str(activity.id)}"] = { 'purpose': activity.reference, 'name': activity.other_name, - prefix: entity, + prefix: entity, # type: ignore[misc] 'amount': activity.amount } field_list.append((str(activity.id), BooleanField(str(activity.id), default=True))) - return field_list, matched_activities + if prefix == "user": + return field_list, t.cast(dict[str, UserMatch], matched_activities) + if prefix == "team": + return field_list, t.cast(dict[str, TeamMatch], matched_activities) + raise ValueError(f"Invalid Prefix: {prefix}") @bp.route('/bank-account-activities/match/do/', methods=['GET', 'POST']) @@ -577,7 +640,10 @@ def bank_account_activities_do_match() -> ResponseReturnValue: matched_team=matched_team) -def _create_field_list(matching): +def _create_field_list( + matching: t.Mapping[BankAccountActivity, User | Account] +) -> list[tuple[str, BooleanField]]: + # TODO use list comprehension field_list = [] for activity, entity in matching.items(): field_list.append( @@ -588,7 +654,10 @@ def _create_field_list(matching): return field_list -def _create_combined_form(field_list_user, field_list_team): +def _create_combined_form( + field_list_user: t.Iterable[tuple[str, Field]], + field_list_team: t.Iterable[tuple[str, Field]], +) -> FlaskForm: form_user = _create_form(field_list_user) form_team = _create_form(field_list_team) @@ -599,7 +668,9 @@ class Form(FlaskForm): return Form() -def _create_form(field_list): +def _create_form( + field_list: t.Iterable[tuple[str, Field]] +) -> type[forms.ActivityMatchForm]: class F(forms.ActivityMatchForm): pass @@ -608,12 +679,14 @@ class F(forms.ActivityMatchForm): return F -def _apply_checked_matches(matching, subform): +def _apply_checked_matches( + matching: UserMatching | AccountMatching, subform: FormField +) -> list[tuple[BankAccountActivity, User | Account]]: # look for all matches which were checked matched = [] for activity, entity in matching.items(): if subform[str(activity.id)].data and activity.transaction_id is None: - debit_account = entity if type(entity) is Account else entity.account + debit_account = entity.account if isinstance(entity, User) else entity credit_account = activity.bank_account.account transaction = finance.simple_transaction( description=activity.reference, @@ -642,7 +715,7 @@ def accounts_list() -> ResponseReturnValue: @bp.route('/account//toggle-legacy') @access.require('finance_change') -def account_toggle_legacy(account_id) -> ResponseReturnValue: +def account_toggle_legacy(account_id: int) -> ResponseReturnValue: account = session.get(Account, account_id) if not account: @@ -658,10 +731,13 @@ def account_toggle_legacy(account_id) -> ResponseReturnValue: @bp.route('/accounts//balance/json') -def balance_json(account_id) -> ResponseReturnValue: +def balance_json(account_id: int) -> ResponseReturnValue: invert = request.args.get('invert', 'False') == 'True' - sum_exp = func.sum(Split.amount).over(order_by=Transaction.valid_on) + sum_exp: ColumnElement[int] = t.cast( + Over[int], + func.sum(Split.amount).over(order_by=Transaction.valid_on), # type: ignore[no-untyped-call] + ) if invert: sum_exp = -sum_exp @@ -679,7 +755,7 @@ def balance_json(account_id) -> ResponseReturnValue: @bp.route('/accounts/') -def accounts_show(account_id) -> ResponseReturnValue: +def accounts_show(account_id: int) -> ResponseReturnValue: account = session.get(Account, account_id) if account is None: @@ -698,12 +774,7 @@ def accounts_show(account_id) -> ResponseReturnValue: inverted = account.type == "USER_ASSET" - _table_kwargs = { - 'data_url': url_for("finance.accounts_show_json", account_id=account_id), - 'saldo': account.balance, - 'inverted': inverted - } - + tbl_data_url = url_for("finance.accounts_show_json", account_id=account_id) balance = -account.balance if inverted else account.balance return render_template( @@ -711,13 +782,23 @@ def accounts_show(account_id) -> ResponseReturnValue: account=account, user=user, balance=balance, balance_json_url=url_for('.balance_json', account_id=account_id, invert=inverted), - finance_table_regular=FinanceTable(**_table_kwargs), - finance_table_splitted=FinanceTableSplitted(**_table_kwargs), + finance_table_regular=FinanceTable( + data_url=tbl_data_url, + saldo=account.balance, + inverted=inverted, + ), + finance_table_splitted=FinanceTableSplitted( + data_url=tbl_data_url, + saldo=account.balance, + inverted=inverted, + ), account_name=localized(account.name, {int: {'insert_commas': False}}) ) -def _format_row(split, style, prefix=None): +def _format_row( + split: Split, style: str, prefix: str | None = None +) -> dict[str, t.Any]: inverted = style == "inverted" row = FinanceRow( posted_at=datetime_filter(split.transaction.posted_at), @@ -746,7 +827,7 @@ def _format_row(split, style, prefix=None): def _prefixed_merge( - a: dict[str, T], prefix_a: str, b: dict[str, U], prefix_b: str + a: t.Mapping[str, T], prefix_a: str, b: t.Mapping[str, U], prefix_b: str ) -> dict[str, T | U]: result: dict[str, T | U] = {} result.update(**{f'{prefix_a}_{k}': v @@ -757,7 +838,7 @@ def _prefixed_merge( @bp.route('/accounts//json') -def accounts_show_json(account_id) -> ResponseReturnValue: +def accounts_show_json(account_id: int) -> ResponseReturnValue: style = request.args.get('style') limit = request.args.get('limit', type=int) offset = request.args.get('offset', type=int) @@ -777,7 +858,7 @@ def accounts_show_json(account_id) -> ResponseReturnValue: sort_order=sort_order, offset=offset, limit=limit, eagerload=True) - def rows_from_query(query): + def rows_from_query(query: Select[tuple[Split]]) -> list[dict[str, t.Any]]: # iterating over `query` executes it return [_format_row(split, style) for split in session.scalars(query)] @@ -802,7 +883,7 @@ def rows_from_query(query): @bp.route('/transactions/') -def transactions_show(transaction_id) -> ResponseReturnValue: +def transactions_show(transaction_id: int) -> ResponseReturnValue: transaction = session.get(Transaction, transaction_id) if transaction is None: @@ -820,7 +901,7 @@ def transactions_show(transaction_id) -> ResponseReturnValue: @bp.route('/transactions//json') -def transactions_show_json(transaction_id) -> ResponseReturnValue: +def transactions_show_json(transaction_id: int) -> ResponseReturnValue: transaction = session.get(Transaction, transaction_id) return TransactionSplitResponse( description=transaction.description, @@ -849,7 +930,9 @@ def transactions_unconfirmed() -> ResponseReturnValue: ) -def _iter_transaction_buttons(bank_acc_act, transaction) -> t.Iterator[BtnColResponse]: +def _iter_transaction_buttons( + bank_acc_act: BankAccountActivity | None, transaction: Transaction +) -> t.Iterator[BtnColResponse]: if not privilege_check(current_user, "finance_change"): return @@ -942,7 +1025,7 @@ def transactions_unconfirmed_json() -> ResponseReturnValue: @bp.route('/transaction//confirm', methods=['GET', 'POST']) @access.require('finance_change') -def transaction_confirm(transaction_id) -> ResponseReturnValue: +def transaction_confirm(transaction_id: int) -> ResponseReturnValue: transaction = session.get(Transaction, transaction_id) if transaction is None: @@ -1018,7 +1101,7 @@ def default_response() -> str: @bp.route('/transaction//delete', methods=['GET', 'POST']) @access.require('finance_change') -def transaction_delete(transaction_id) -> ResponseReturnValue: +def transaction_delete(transaction_id: int) -> ResponseReturnValue: transaction = session.get(Transaction, transaction_id) if transaction is None: @@ -1157,7 +1240,7 @@ def accounts_create() -> ResponseReturnValue: @bp.route("/membership_fee//book", methods=['GET', 'POST']) @access.require('finance_change') -def membership_fee_book(fee_id) -> ResponseReturnValue: +def membership_fee_book(fee_id: int) -> ResponseReturnValue: fee = session.get(MembershipFee, fee_id) if fee is None: @@ -1181,7 +1264,7 @@ def membership_fee_book(fee_id) -> ResponseReturnValue: @bp.route("/membership_fee//users_due_json") -def membership_fee_users_due_json(fee_id) -> ResponseReturnValue: +def membership_fee_users_due_json(fee_id: int) -> ResponseReturnValue: fee = session.get(MembershipFee, fee_id) if fee is None: @@ -1315,7 +1398,7 @@ def membership_fee_create() -> ResponseReturnValue: @bp.route('/membership_fee//edit', methods=("GET", "POST")) @access.require('finance_change') -def membership_fee_edit(fee_id) -> ResponseReturnValue: +def membership_fee_edit(fee_id: int) -> ResponseReturnValue: fee = session.get(MembershipFee, fee_id) if fee is None: diff --git a/web/blueprints/finance/forms.py b/web/blueprints/finance/forms.py index 1b3008022..7eb880094 100644 --- a/web/blueprints/finance/forms.py +++ b/web/blueprints/finance/forms.py @@ -4,9 +4,11 @@ import datetime from flask_wtf import FlaskForm as Form -from wtforms import Form as WTForm, ValidationError +from wtforms import Form as WTForm, ValidationError, Field from wtforms.validators import DataRequired, NumberRange, Optional, \ InputRequired + +from pycroft.model.user import User from wtforms_widgets.fields.core import ( TextField, IntegerField, HiddenField, FileField, SelectField, FormField, FieldList, StringField, DateField, MoneyField, PasswordField, BooleanField, @@ -82,7 +84,7 @@ class BankAccountCreateForm(Form): bic = TextField("BIC") fints = TextField("FinTS-Endpunkt", default="https://mybank.com/…") - def validate_iban(self, field): + def validate_iban(self, field: Field) -> None: if BankAccount.q.filter_by(iban=field.data ).first() is not None: raise ValidationError("Konto existiert bereits.") @@ -138,7 +140,7 @@ class SplitCreateForm(WTForm): account_id = HiddenField(validators=[DataRequired(message=gettext("Missing account."))]) amount = MoneyField("Wert", validators=[DataRequired(message=gettext("Invalid value."))]) - def validate_amount(self, field): + def validate_amount(self, field: Field) -> None: cents = field.data.shift(2) if cents == 0 or cents != int(cents): raise ValidationError(gettext("Invalid value.")) @@ -155,7 +157,7 @@ class TransactionCreateForm(Form): min_entries=2 ) - def validate_splits(self, field): + def validate_splits(self, field: FormField) -> None: balance = sum(split_form['amount'].data for split_form in field if split_form['amount'].data is not None) if balance != 0: @@ -166,11 +168,11 @@ class ActivityMatchForm(Form): pass -def get_user_name_with_id(user): +def get_user_name_with_id(user: User) -> str: return f"{user.name} ({user.id})" -def get_user_name_with_id_and_balance(user): +def get_user_name_with_id_and_balance(user: User) -> str: return f"{user.name} ({user.id}) | {-user.account.balance}€" From 9de9df8388af79ea50d71d9a6ffe517a1cd22337 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:18:12 +0200 Subject: [PATCH 09/18] [web][typing] type hint `json_agg_core` and remove `json_agg` The latter was unused. --- web/blueprints/helpers/api.py | 10 +++------- 1 file changed, 3 insertions(+), 7 deletions(-) diff --git a/web/blueprints/helpers/api.py b/web/blueprints/helpers/api.py index f3f0324bc..39671feef 100644 --- a/web/blueprints/helpers/api.py +++ b/web/blueprints/helpers/api.py @@ -1,15 +1,11 @@ # Copyright (c) 2016 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 pycroft.model import session -from sqlalchemy.sql import func, Alias, literal_column, select +from sqlalchemy import Select, SelectBase +from sqlalchemy.sql import func, literal_column, select -def json_agg(query): - return session.session.query(func.json_agg(literal_column("row"))) \ - .select_from(Alias(query.subquery(), "row")) - -def json_agg_core(selectable): +def json_agg_core(selectable: SelectBase) -> Select[tuple[list[dict]]]: return select(func.json_agg(literal_column("row"))) \ .select_from(selectable.alias("row")) From 53513055ab8c784f5a5ccb49250440e9fc506793 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:22:14 +0200 Subject: [PATCH 10/18] [web][typing] improve typing of `@lazy_join` properties on tables --- web/blueprints/facilities/tables.py | 7 ++++--- web/blueprints/finance/tables.py | 14 +++++++------ web/blueprints/host/tables.py | 7 ++++--- web/blueprints/user/tables.py | 7 ++++--- web/table/lazy_join.py | 31 +++++++++++++++++++++-------- web/table/table.py | 16 +++++++++------ 6 files changed, 53 insertions(+), 29 deletions(-) diff --git a/web/blueprints/facilities/tables.py b/web/blueprints/facilities/tables.py index 49072215c..d7f37c611 100644 --- a/web/blueprints/facilities/tables.py +++ b/web/blueprints/facilities/tables.py @@ -1,6 +1,7 @@ from flask import url_for from pydantic import BaseModel +from web.table.lazy_join import LazilyJoined from web.table.table import ( BootstrapTable, Column, @@ -37,7 +38,7 @@ class Meta: inhabitants = MultiBtnColumn('Bewohner') @property - def toolbar(self): + def toolbar(self) -> LazilyJoined: return toggle_button_toolbar( "Display all users", id="rooms-toggle-all-users", @@ -88,9 +89,9 @@ def __init__(self, *a, room_id=None, **kw) -> None: self.room_id = room_id @property - def toolbar(self): + def toolbar(self) -> LazilyJoined | None: if no_inf_change(): - return + return None href = url_for(".patch_port_create", switch_room_id=self.room_id) return button_toolbar("Patch-Port", href) diff --git a/web/blueprints/finance/tables.py b/web/blueprints/finance/tables.py index c8cfe9d19..85b94407a 100644 --- a/web/blueprints/finance/tables.py +++ b/web/blueprints/finance/tables.py @@ -1,4 +1,5 @@ import typing +import typing as t from decimal import Decimal from flask import url_for @@ -6,6 +7,7 @@ from flask_login import current_user from pydantic import BaseModel +from web.table.lazy_join import HasDunderStr, LazilyJoined from web.table.table import ( lazy_join, DictValueMixin, @@ -76,20 +78,20 @@ def __init__(self, *a, saldo=None, user_id=None, inverted=False, **kw) -> None: amount = ColoredColumn("Wert", cell_style='table.tdRelativeCellStyle') @property - def toolbar(self): + def toolbar(self) -> HasDunderStr | None: """Generate a toolbar with a details button If a user_id was passed in the constructor, this renders a “details” button reaching the finance overview of the user's account. """ if self.user_id is None: - return + return None href = url_for("user.user_account", user_id=self.user_id) return button_toolbar("Details", href, icon="fa-chart-area") @property @lazy_join - def table_footer(self): + def table_footer(self) -> t.Iterator[str]: yield "" yield "" @@ -144,7 +146,7 @@ class MembershipFeeTable(BootstrapTable): actions = MultiBtnColumn("Aktionen") @property - def toolbar(self): + def toolbar(self) -> LazilyJoined: """An “add fee” button""" href = url_for(".membership_fee_create") return button_toolbar(gettext("Beitrag erstellen"), href) @@ -198,10 +200,10 @@ def __init__(self, *a, create_account=False, **kw): super().__init__(*a, **kw) @property - def toolbar(self): + def toolbar(self) -> HasDunderStr | None: """A “create bank account” button""" if not self.create_account: - return + return None href = url_for(".bank_accounts_create") return button_toolbar(gettext("Neues Bankkonto anlegen"), href) diff --git a/web/blueprints/host/tables.py b/web/blueprints/host/tables.py index 0252c917f..82d824563 100644 --- a/web/blueprints/host/tables.py +++ b/web/blueprints/host/tables.py @@ -2,6 +2,7 @@ from pydantic import BaseModel from web.blueprints.helpers.user import no_hosts_change +from web.table.lazy_join import HasDunderStr from web.table.table import ( BootstrapTable, Column, @@ -34,11 +35,11 @@ def __init__(self, *a, user_id: int | None = None, **kw) -> None: self.user_id = user_id @property - def toolbar(self): + def toolbar(self) -> HasDunderStr | None: if self.user_id is None: - return + return None if no_hosts_change(): - return + return None href = url_for("host.host_create", user_id=self.user_id) return button_toolbar("Host", href) diff --git a/web/blueprints/user/tables.py b/web/blueprints/user/tables.py index b057682d4..c3c920641 100644 --- a/web/blueprints/user/tables.py +++ b/web/blueprints/user/tables.py @@ -4,6 +4,7 @@ from pydantic import BaseModel from web.blueprints.helpers.log_tables import RefreshableTableMixin +from web.table.lazy_join import HasDunderStr from web.table.table import ( BootstrapTable, Column, @@ -35,11 +36,11 @@ def __init__(self, *a, user_id: int | None = None, **kw) -> None: self.user_id = user_id @property - def toolbar(self): + def toolbar(self) -> HasDunderStr | None: if self.user_id is None: - return + return None if no_membership_change(): - return + return None href = url_for(".add_membership", user_id=self.user_id) return button_toolbar("Mitgliedschaft", href) diff --git a/web/table/lazy_join.py b/web/table/lazy_join.py index 5dbd85be6..e81cb93e0 100644 --- a/web/table/lazy_join.py +++ b/web/table/lazy_join.py @@ -1,6 +1,7 @@ +import typing as t from functools import wraps from types import FunctionType -from typing import Generator, Callable, Iterable, overload +from typing import Generator, Callable, overload def filled_iter(iter, filler): @@ -14,33 +15,45 @@ def filled_iter(iter, filler): yield elem +class HasDunderStr(t.Protocol): + def __str__(self) -> str: + ... + + class LazilyJoined: """A string that consists of multiple components NOTE: Just like a generator, it will be exhausted after the first call! """ - _components: Iterable - def __init__(self, components: Iterable, glue: str = ""): + glue: str + _components: t.Iterable[HasDunderStr | None] + exhausted: bool + + def __init__( + self, + components: t.Iterable[HasDunderStr | None], + glue: str = "", + ): self.glue = glue self._components = components self.exhausted = False @property - def _stringified_components(self): + def _stringified_components(self) -> t.Iterator[str]: return ((str(c) if c is not None else "") for c in self._components) - def __str__(self): + def __str__(self) -> str: self._mark_exhausted() return self.glue.join(self._stringified_components) - def __iter__(self): + def __iter__(self) -> t.Iterator[str]: self._mark_exhausted() if self.glue: return filled_iter(self._stringified_components, filler=self.glue) return iter(self._stringified_components) - def _mark_exhausted(self): + def _mark_exhausted(self) -> None: if self.exhausted: raise RuntimeError("LazyJoined object already exhausted!" " You may call __str__ or __iter__ only once." @@ -48,7 +61,9 @@ def _mark_exhausted(self): self.exhausted = True -DecoratedInType = Callable[..., (Generator[str, None, None])] +DecoratedInType = Callable[ + ..., Generator[HasDunderStr | None, None, None] | t.Iterator[HasDunderStr | None] +] DecoratedOutType = Callable[..., LazilyJoined] @overload diff --git a/web/table/table.py b/web/table/table.py index 4ee7890c3..9874f983b 100644 --- a/web/table/table.py +++ b/web/table/table.py @@ -12,7 +12,7 @@ from pydantic import BaseModel, Field from annotated_types import Predicate -from .lazy_join import lazy_join, LazilyJoined +from .lazy_join import lazy_join, LazilyJoined, HasDunderStr class Column: @@ -301,7 +301,7 @@ class IbanColumn(Column): UnboundTableArgs = frozenset[tuple[str, Any]] -TableArgs = dict[str, str] +TableArgs = dict[HasDunderStr, HasDunderStr] def _infer_table_args(meta_obj, superclass_table_args: TableArgs) -> UnboundTableArgs: @@ -419,19 +419,23 @@ def _init_table_args(self): @property @lazy_join - def table_header(self): + def table_header(self) -> t.Iterator[HasDunderStr]: yield "" yield "" yield from self.columns yield "" yield "" - toolbar = "" + @property + def toolbar(self) -> HasDunderStr | None: + return "" - table_footer = "" + @property + def table_footer(self) -> HasDunderStr | None: + return "" @lazy_join("\n") - def _render(self, table_id): + def _render(self, table_id) -> t.Iterator[HasDunderStr | None]: toolbar_args = html_params(id=f"{table_id}-toolbar", class_="btn-toolbar", role="toolbar") From a094278479c7a05b2451ecd20eda7366bb92d9e8 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:23:42 +0200 Subject: [PATCH 11/18] [web][typing] add remaining type hints for `finance` bp --- pycroft/lib/finance/transaction_crud.py | 2 +- web/blueprints/finance/tables.py | 32 +++++++++++++++---------- 2 files changed, 20 insertions(+), 14 deletions(-) diff --git a/pycroft/lib/finance/transaction_crud.py b/pycroft/lib/finance/transaction_crud.py index 4437d8645..422026b47 100644 --- a/pycroft/lib/finance/transaction_crud.py +++ b/pycroft/lib/finance/transaction_crud.py @@ -30,7 +30,7 @@ def simple_transaction( description: str, debit_account: Account, credit_account: Account, - amount: Decimal, + amount: Decimal | int, author: User, valid_on: date | None = None, confirmed: bool = True, diff --git a/web/blueprints/finance/tables.py b/web/blueprints/finance/tables.py index 85b94407a..ed2ab1c93 100644 --- a/web/blueprints/finance/tables.py +++ b/web/blueprints/finance/tables.py @@ -47,13 +47,19 @@ class Meta: 'data-page-list': '[5, 10, 25, 50, 100]' } - def __init__(self, *a, saldo=None, user_id=None, inverted=False, **kw) -> None: + def __init__( + self, + *, + saldo: int | Decimal | None = None, + user_id: int | None = None, + inverted: bool = False, + **kw: t.Any, + ) -> None: """Init - :param int user_id: An optional user_id. If set, this causes - a “details” button to be rendered in the toolbar - referencing the user. - :param bool inverted: An optional switch adding + :param user_id: If set, this causes a “details” button + to be rendered in the toolbar referencing the user. + :param inverted: An optional switch adding `style=inverted` to the given `data_url` """ @@ -66,7 +72,7 @@ def __init__(self, *a, saldo=None, user_id=None, inverted=False, **kw) -> None: ) self.saldo = -saldo - super().__init__(*a, **kw) + super().__init__(**kw) self.user_id = user_id @@ -126,12 +132,12 @@ class Meta: splits = (('soll', "Soll"), ('haben', "Haben")) - def __init__(self, *a, **kw) -> None: - super().__init__(*a, **kw) + def __init__(self, **kw: t.Any) -> None: + super().__init__(**kw) self.table_footer_offset = 7 -def no_finance_change(): +def no_finance_change() -> bool: return not current_user.has_property('finance_change') @@ -195,9 +201,9 @@ class BankAccountTable(BootstrapTable): last_imported_at = Column("Zuletzt importiert") kto = BtnColumn("Konto") - def __init__(self, *a, create_account=False, **kw): + def __init__(self, *, create_account: bool = False, **kw: t.Any) -> None: self.create_account = create_account - super().__init__(*a, **kw) + super().__init__(**kw) @property def toolbar(self) -> HasDunderStr | None: @@ -230,14 +236,14 @@ class BankAccountActivityTable(BootstrapTable): amount = Column("Betrag", width=1, formatter="table.euroFormatter") actions = MultiBtnColumn("Aktionen", width=1) - def __init__(self, *a, **kw): + def __init__(self, **kw: t.Any) -> None: table_args = kw.pop('table_args', {}) table_args.setdefault('data-detail-view', "true") table_args.setdefault('data-row-style', "table.financeRowFormatter") table_args.setdefault('data-detail-formatter', "table.bankAccountActivitiesDetailFormatter") kw['table_args'] = table_args - super().__init__(*a, **kw) + super().__init__(**kw) class Meta: table_args = { From 6158b434f0e7d750392ebb8a02290acd2075826f Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:24:40 +0200 Subject: [PATCH 12/18] [web][typing] add remaining typehints for `user` blueprint --- pycroft/lib/user.py | 3 ++- pycroft/model/user.py | 2 +- web/blueprints/helpers/user.py | 8 ++++---- web/blueprints/user/__init__.py | 29 +++++++++++++++++------------ web/blueprints/user/forms.py | 21 ++++++++++++--------- web/blueprints/user/log.py | 26 +++++++++++++------------- web/blueprints/user/tables.py | 5 +++-- 7 files changed, 52 insertions(+), 42 deletions(-) diff --git a/pycroft/lib/user.py b/pycroft/lib/user.py index 38ab0286c..e8367cff8 100644 --- a/pycroft/lib/user.py +++ b/pycroft/lib/user.py @@ -919,11 +919,12 @@ def generate_user_sheet( Only necessary if wifi=True: :param generation_purpose: Optional purpose why this usersheet was printed """ + from pycroft.helpers import printing return generate_pdf( new_user=new_user, wifi=wifi, bank_account=config.membership_fee_bank_account, - user=user, + user=t.cast(printing.User, user), user_id=encode_type2_user_id(user.id), plain_user_password=plain_user_password, generation_purpose=generation_purpose, diff --git a/pycroft/model/user.py b/pycroft/model/user.py index acaa98f0f..827b39bcb 100644 --- a/pycroft/model/user.py +++ b/pycroft/model/user.py @@ -373,7 +373,7 @@ def has_wifi_access(self): return self.wifi_passwd_hash is not None @staticmethod - def verify_and_get(login, plaintext_password): + def verify_and_get(login: str, plaintext_password: str) -> User | None: try: user = User.q.filter(User.login == func.lower(login)).one() except NoResultFound: diff --git a/web/blueprints/helpers/user.py b/web/blueprints/helpers/user.py index e726b6ecc..c3c645915 100644 --- a/web/blueprints/helpers/user.py +++ b/web/blueprints/helpers/user.py @@ -3,10 +3,10 @@ from pycroft.model.session import session from pycroft.model.user import User, PreMember -from web.table.table import BtnColResponse +from web.table.table import BtnColResponse, BtnClass, IconClass -def user_btn_style(user): +def user_btn_style(user: User) -> tuple[BtnClass, list[IconClass], str]: """Determine the icons and style of the button to a users page. First, add glyphicons concerning status warnings (finance, @@ -93,9 +93,9 @@ def get_pre_member_or_404(prm_id: int) -> PreMember: return prm -def no_membership_change(): +def no_membership_change() -> bool: return not current_user.has_property('groups_change_membership') -def no_hosts_change(): +def no_hosts_change() -> bool: return not current_user.has_property('hosts_change') diff --git a/web/blueprints/user/__init__.py b/web/blueprints/user/__init__.py index 21624e408..abbc1c0c2 100644 --- a/web/blueprints/user/__init__.py +++ b/web/blueprints/user/__init__.py @@ -12,6 +12,7 @@ import re import typing as t from datetime import timedelta +from decimal import Decimal from functools import partial from typing import TypeVar, Callable, cast @@ -306,14 +307,7 @@ def user_show(user_id: int) -> ResponseReturnValue: balance = user.account.balance _log_endpoint = partial(url_for, ".user_show_logs_json", user_id=user.id) _membership_endpoint = partial(url_for, ".user_show_groups_json", user_id=user.id) - _finance_table_kwargs = { - 'data_url': url_for("finance.accounts_show_json", account_id=user.account_id), - 'user_id': user.id, - 'table_args': {'data-page-size': 5}, - 'inverted': True, - 'saldo': balance, - } - + tbl_data_url = url_for("finance.accounts_show_json", account_id=user.account_id) is_blocked = False for group in get_blocked_groups(): @@ -356,13 +350,24 @@ def user_show(user_id: int) -> ResponseReturnValue: user_id=user.id), task_table=TaskTable(data_url=url_for("task.json_tasks_for_user", user_id=user.id), hidden_columns=['user']), - finance_table_regular=FinanceTable(**_finance_table_kwargs), - finance_table_splitted=FinanceTableSplitted(**_finance_table_kwargs), + finance_table_regular=FinanceTable( + data_url=tbl_data_url, + user_id=user.id, + table_args={"data-page-size": 5}, + inverted=True, + saldo=balance, + ), + finance_table_splitted=FinanceTableSplitted( + data_url=tbl_data_url, + user_id=user.id, + table_args={"data-page-size": 5}, + inverted=True, + saldo=balance, + ), room=room, form=form, flags=infoflags(user), - json_url=url_for("finance.accounts_show_json", - account_id=user.account_id), + json_url=tbl_data_url, is_blocked=is_blocked, granted_properties=sorted(p.property_name for p in user.current_properties), revoked_properties=sorted( diff --git a/web/blueprints/user/forms.py b/web/blueprints/user/forms.py index 8739d5b3e..55708c359 100644 --- a/web/blueprints/user/forms.py +++ b/web/blueprints/user/forms.py @@ -6,6 +6,7 @@ from flask_wtf import FlaskForm as Form from markupsafe import escape, Markup from sqlalchemy import select +from sqlalchemy.orm import Query from wtforms import Field from wtforms.validators import ( Regexp, ValidationError, DataRequired, Email, Optional) @@ -35,19 +36,19 @@ iter_prefixed_field_names -def user_query(): +def user_query() -> Query: return User.q.order_by(User.id) -def host_query(): +def host_query() -> Query: return Host.q.order_by(Host.id) -def group_query(): +def group_query() -> Query: return PropertyGroup.q.order_by(PropertyGroup.name) -def validate_unique_login(form, field): +def validate_unique_login(form: Form, field: Field) -> None: if User.q.filter_by(login=field.data).first(): raise ValidationError("Nutzerlogin schon vergeben!") @@ -60,7 +61,8 @@ class UniqueName: :param force_field: The name of the “do it anyway” checkbox field """ - def __init__(self, force_field: str | None = 'force'): + + def __init__(self, force_field: str | None = "force") -> None: self.force_field = force_field self.building_field = 'building' self.level_field = 'level' @@ -87,7 +89,7 @@ def try_get_room(self, form: Form) -> Room | None: select(Room).filter_by(number=number, level=level, building=building) ) - def __call__(self, form: Form, field: Field): + def __call__(self, form: Form, field: Field) -> None: if self.force_set(form): return if (room := self.try_get_room(form)) is None: @@ -115,7 +117,8 @@ class UniqueEmail: :param force_field: The name of the “do it anyway” checkbox field """ - def __init__(self, force_field: str | None = 'force'): + + def __init__(self, force_field: str | None = "force") -> None: self.force_field = force_field def force_set(self, form: Form) -> bool: @@ -129,7 +132,7 @@ def force_set(self, form: Form) -> bool: def get_conflicting_users(email: str) -> list[User]: return User.q.filter_by(email=email).all() - def __call__(self, form: Form, field: Field): + def __call__(self, form: Form, field: Field) -> None: if self.force_set(form): return if not (conflicting_users := self.get_conflicting_users(field.data)): @@ -182,7 +185,7 @@ class UserEditForm(Form): class UserEditAddressForm(CreateAddressForm): - def set_defaults_from_adress(self, address: Address): + def set_defaults_from_adress(self, address: Address) -> None: self.address_street.data = address.street self.address_number.data = address.number self.address_zip_code.data = address.zip_code diff --git a/web/blueprints/user/log.py b/web/blueprints/user/log.py index 9697b9056..889197887 100644 --- a/web/blueprints/user/log.py +++ b/web/blueprints/user/log.py @@ -7,47 +7,47 @@ import logging import typing as t +from ipaddr import _BaseIP +from sqlalchemy import select, Row from sqlalchemy.orm import Query -from hades_logs import hades_logs +from hades_logs import hades_logs, RadiusLogEntry from hades_logs.exc import HadesConfigError, HadesOperationalError, HadesTimeout from pycroft.model import session -from pycroft.model.host import SwitchPort +from pycroft.model.host import SwitchPort, Switch from pycroft.model.port import PatchPort from pycroft.model.user import User +from pycroft.model.facilities import Room from ..helpers.log import format_hades_log_entry, format_hades_disabled, \ format_user_not_connected, format_hades_error, format_hades_timeout from ..helpers.log_tables import LogTableRow logger = logging.getLogger(__name__) -def iter_hades_switch_ports(room): +def iter_hades_switch_ports(room: Room) -> t.Sequence[Row[tuple[str, _BaseIP]]]: """Return all tuples of (nasportid, nasipaddress) for a room. - :param Room room: The room to filter by + :param room: The room to filter by :returns: An iterator of (nasportid, nasipaddress) usable as arguments in ``HadesLogs.fetch_logs()`` """ - from pycroft.model._all import Room, PatchPort, SwitchPort, Switch - query = ( - session.session - .query(SwitchPort.name, Switch.management_ip) + stmt = ( + select(SwitchPort.name, Switch.management_ip) .join(PatchPort.room) .join(PatchPort.switch_port) .join(SwitchPort.switch) .filter(Room.id == room.id) ) - return query.all() + return session.session.execute(stmt).all() -def get_user_hades_logs(user): +def get_user_hades_logs(user: User) -> t.Iterator[tuple[PatchPort, RadiusLogEntry]]: """Iterate over a user's hades logs - :param User user: the user whose logs to display + :param user: the user whose logs to display :returns: an iterator over duples (interface, log_entry). - :rtype: Iterator[SwitchPort, RadiusLogEntry] """ # Accessing the `hades_logs` proxy early ensures the exception is # raised even if there's no SwitchPort @@ -67,7 +67,7 @@ def get_user_hades_logs(user): yield port, log_entry -def is_user_connected(user): +def is_user_connected(user: User) -> bool: try: next(patch_port for host in user.hosts diff --git a/web/blueprints/user/tables.py b/web/blueprints/user/tables.py index c3c920641..f17fb4bf4 100644 --- a/web/blueprints/user/tables.py +++ b/web/blueprints/user/tables.py @@ -1,3 +1,4 @@ +import typing as t import typing from flask import url_for @@ -31,8 +32,8 @@ class MembershipTable(BootstrapTable): ends_at = DateColumn("Ende") actions = MultiBtnColumn("Aktionen", hide_if=no_membership_change) - def __init__(self, *a, user_id: int | None = None, **kw) -> None: - super().__init__(*a, **kw) + def __init__(self, *, user_id: int | None = None, **kw: t.Any) -> None: + super().__init__(**kw) self.user_id = user_id @property From 964d2d38f1b389dfd73f5fc2b2a0742967249d08 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:25:56 +0200 Subject: [PATCH 13/18] [web][typing] add remaining type hints to `table`s Additionaly, remove the variadic positional `*a` argument, which only makes typing more confusing and is never really used. --- web/blueprints/facilities/tables.py | 6 ++++-- web/blueprints/helpers/log_tables.py | 10 ++++++---- web/blueprints/host/tables.py | 10 ++++++---- web/blueprints/infrastructure/tables.py | 17 ++++++++++------- web/blueprints/task/tables.py | 8 ++++++-- web/table/table.py | 16 ++++++++-------- 6 files changed, 40 insertions(+), 27 deletions(-) diff --git a/web/blueprints/facilities/tables.py b/web/blueprints/facilities/tables.py index d7f37c611..75d0fe0ec 100644 --- a/web/blueprints/facilities/tables.py +++ b/web/blueprints/facilities/tables.py @@ -1,3 +1,5 @@ +import typing as t + from flask import url_for from pydantic import BaseModel @@ -83,8 +85,8 @@ class Meta: edit_link = BtnColumn('Editieren', hide_if=no_inf_change) delete_link = BtnColumn('Löschen', hide_if=no_inf_change) - def __init__(self, *a, room_id=None, **kw) -> None: - super().__init__(*a, **kw) + def __init__(self, *, room_id: int | None = None, **kw: t.Any) -> None: + super().__init__(**kw) self.room_id = room_id diff --git a/web/blueprints/helpers/log_tables.py b/web/blueprints/helpers/log_tables.py index 7eb3a4d66..74f875f41 100644 --- a/web/blueprints/helpers/log_tables.py +++ b/web/blueprints/helpers/log_tables.py @@ -13,21 +13,23 @@ UserColumn, DateColResponse, UserColResponse, + TableArgs, ) -class RefreshableTableMixin: +class RefreshableTableMixin(BootstrapTable): """A mixin class showing the refresh button by default. In :py:meth:`__init__`s ``table_args`` argument, a default of ``{'data-show-refresh': "true"}`` is established. """ - def __init__(self, *a, **kw) -> None: - table_args = kw.pop("table_args", {}) + def __init__(self, *, table_args: TableArgs | None = None, **kw: t.Any) -> None: + if table_args is None: + table_args = {} table_args.setdefault("data-show-refresh", "true") kw["table_args"] = table_args - super().__init__(*a, **kw) + super().__init__(**kw) class LogTableExtended(RefreshableTableMixin, BootstrapTable): diff --git a/web/blueprints/host/tables.py b/web/blueprints/host/tables.py index 82d824563..44d9713c4 100644 --- a/web/blueprints/host/tables.py +++ b/web/blueprints/host/tables.py @@ -1,3 +1,5 @@ +import typing as t + from flask import url_for from pydantic import BaseModel @@ -23,7 +25,7 @@ class HostTable(BootstrapTable): interface_create_link = Column("", hide_if=lambda: True) id = Column("", hide_if=lambda: True) - def __init__(self, *a, user_id: int | None = None, **kw) -> None: + def __init__(self, *, user_id: int | None = None, **kw: t.Any) -> None: table_args = kw.pop('table_args', {}) table_args.setdefault('data-load-subtables', "true") table_args.setdefault('data-detail-view', "true") @@ -31,7 +33,7 @@ def __init__(self, *a, user_id: int | None = None, **kw) -> None: table_args.setdefault('data-response-handler', "table.userHostResponseHandler") kw['table_args'] = table_args - super().__init__(*a, **kw) + super().__init__(**kw) self.user_id = user_id @property @@ -63,13 +65,13 @@ class InterfaceTable(BootstrapTable): ips = Column("IPs") actions = MultiBtnColumn("Aktionen", hide_if=no_hosts_change) - def __init__(self, *a, host_id: int | None = None, **kw) -> None: + def __init__(self, *, host_id: int | None = None, **kw: t.Any) -> None: table_args = kw.pop("table_args", {}) table_args.setdefault("data-hide-pagination-info", "true") table_args.setdefault("data-search", "false") kw["table_args"] = table_args - super().__init__(*a, **kw) + super().__init__(**kw) self.host_id = host_id diff --git a/web/blueprints/infrastructure/tables.py b/web/blueprints/infrastructure/tables.py index 2d81fb84b..60bdf7c89 100644 --- a/web/blueprints/infrastructure/tables.py +++ b/web/blueprints/infrastructure/tables.py @@ -1,7 +1,10 @@ +import typing as t + from flask import url_for from flask_login import current_user from pydantic import BaseModel, ConfigDict +from web.table.lazy_join import HasDunderStr from web.table.table import ( BootstrapTable, Column, @@ -13,7 +16,7 @@ ) -def no_inf_change(): +def no_inf_change() -> bool: return not current_user.has_property('infrastructure_change') @@ -48,9 +51,9 @@ class SwitchTable(BootstrapTable): delete_link = BtnColumn('Löschen', width=1, hide_if=no_inf_change) @property - def toolbar(self): + def toolbar(self) -> HasDunderStr | None: if not current_user.has_property('infrastructure_change'): - return + return None return button_toolbar("Switch", url_for(".switch_create")) @@ -82,8 +85,8 @@ class Meta: 'data-sort-name': 'switchport_name', } - def __init__(self, *a, switch_id=None, **kw) -> None: - super().__init__(*a, **kw) + def __init__(self, *, switch_id: int | None = None, **kw: t.Any) -> None: + super().__init__(**kw) self.switch_id = switch_id switchport_name = Column("Name", width=2, col_args={'data-sorter': 'table.sortPort'}) @@ -93,9 +96,9 @@ def __init__(self, *a, switch_id=None, **kw) -> None: delete_link = BtnColumn('Löschen', hide_if=no_inf_change) @property - def toolbar(self): + def toolbar(self) -> HasDunderStr | None: if no_inf_change(): - return + return None href = url_for(".switch_port_create", switch_id=self.switch_id) return button_toolbar("Switch-Port", href) diff --git a/web/blueprints/task/tables.py b/web/blueprints/task/tables.py index dbefc1752..53b863ee3 100644 --- a/web/blueprints/task/tables.py +++ b/web/blueprints/task/tables.py @@ -30,7 +30,11 @@ class TaskTable(BootstrapTable): type = Column("Typ", hide_if=lambda: True) def __init__( - self, hidden_columns: t.Container[str] = None, sort_order="asc", *a, **kw + self, + *, + hidden_columns: t.Container[str] = None, + sort_order: str = "asc", + **kw: t.Any ): table_args = kw.pop("table_args", {}) table_args.setdefault("data-detail-view", "true") @@ -46,7 +50,7 @@ def __init__( kw['table_args'] = table_args - super().__init__(*a, **kw) + super().__init__(**kw) class TaskRow(BaseModel): diff --git a/web/table/table.py b/web/table/table.py index 9874f983b..54a0e49c3 100644 --- a/web/table/table.py +++ b/web/table/table.py @@ -59,7 +59,7 @@ def __init__( sortable=True, hide_if: Callable[[], bool] | None = lambda: False, escape: bool | None = None, - ): + ) -> None: self.name = name self.title = title self.formatter = formatter if formatter is not None else False @@ -138,7 +138,7 @@ class DictListValueMixin: # noinspection PyPep8Naming class custom_formatter_column: - def __init__(self, formatter_name: str): + def __init__(self, formatter_name: str) -> None: self.formatter_name = formatter_name def __call__(self, cls): @@ -168,7 +168,7 @@ class BtnColResponse(BaseModel): @custom_formatter_column('table.btnFormatter') class BtnColumn(DictValueMixin, Column): - def __init__(self, *a, **kw): + def __init__(self, *a, **kw) -> None: super().__init__(*a, sortable=False, **kw) if typing.TYPE_CHECKING: @@ -190,7 +190,7 @@ def value( @custom_formatter_column('table.multiBtnFormatter') class MultiBtnColumn(DictListValueMixin, Column): - def __init__(self, *a, **kw): + def __init__(self, *a, **kw) -> None: super().__init__(*a, sortable=False, **kw) if typing.TYPE_CHECKING: @@ -380,7 +380,7 @@ class BootstrapTable(metaclass=BootstrapTableMeta): class Meta: table_args = {'data-toggle': "table", "data-icons-prefix": "fa"} - def __init__(self, data_url, table_args=None) -> None: + def __init__(self, *, data_url: str, table_args: TableArgs | None = None) -> None: self.data_url = enforce_url_params(data_url, dict(self._enforced_url_params)) # un-freeze the classes table args so it can be modified on the instance self.table_args = dict(self._table_args) @@ -397,11 +397,11 @@ def _columns(self) -> list[Column]: return [getattr(self, a) for a in self.column_attrname_map.values()] @property - def columns(self): + def columns(self) -> t.Sequence[Column]: """Wrapper for subclasses to override.""" return self._columns - def __repr__(self): + def __repr__(self) -> str: return "<{cls} cols={numcols} data_url={data_url!r}>".format( cls=type(self).__name__, numcols=len(self.columns), @@ -597,7 +597,7 @@ def toggle_button_toolbar(title: str, id: str, icon: str = "fa-plus") \ yield "" -def html_params(**kwargs): +def html_params(**kwargs) -> str: """ Generate HTML attribute syntax from inputted keyword arguments. """ From 1ff957e833678b50bb34a703a772e3f9dd55ea47 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:27:25 +0200 Subject: [PATCH 14/18] [web][typing] fix mypy admonitions in remaining, non-bp modules --- web/blueprints/facilities/address.py | 9 ++++++--- web/blueprints/facilities/forms.py | 15 +++++++++------ web/blueprints/helpers/form.py | 15 +++++++++++---- web/blueprints/helpers/host.py | 2 +- web/blueprints/helpers/log.py | 17 ++++++++++++++--- web/blueprints/infrastructure/forms.py | 3 ++- 6 files changed, 43 insertions(+), 18 deletions(-) diff --git a/web/blueprints/facilities/address.py b/web/blueprints/facilities/address.py index 56924189b..3a5b60d1d 100644 --- a/web/blueprints/facilities/address.py +++ b/web/blueprints/facilities/address.py @@ -1,4 +1,4 @@ -from sqlalchemy.orm import Session +from sqlalchemy.orm import Session, Query from sqlalchemy.orm.attributes import InstrumentedAttribute from pycroft.model.address import Address @@ -24,8 +24,11 @@ def get_address_entity(type: str) -> InstrumentedAttribute: ) from None -def address_entity_search_query(query: str, entity: InstrumentedAttribute, session: Session, limit: int): - return (session.query() +def address_entity_search_query( + query: str, entity: InstrumentedAttribute, session: Session, limit: int +) -> Query: + return ( + session.query() .add_columns(entity) .distinct() .filter(entity.ilike(f"%{query}%")) diff --git a/web/blueprints/facilities/forms.py b/web/blueprints/facilities/forms.py index d1525c001..7f9792f28 100644 --- a/web/blueprints/facilities/forms.py +++ b/web/blueprints/facilities/forms.py @@ -1,6 +1,7 @@ # Copyright (c) 2015 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 typing import Callable from flask import url_for @@ -21,14 +22,16 @@ class LazyString: - def __init__(self, value_factory: Callable[[], str]): + def __init__(self, value_factory: Callable[[], str]) -> None: self.value_factory = value_factory - def __str__(self): + def __str__(self) -> str: return self.value_factory() -def create_address_field(name: str, *args, type: str, render_kw: dict | None = None, **kwargs): +def create_address_field( + name: str, *args: t.Any, type: str, render_kw: dict | None = None, **kwargs: t.Any +) -> TypeaheadField: assert type in ADDRESS_ENTITIES, "Unknown address_type" return TypeaheadField( name, @@ -57,11 +60,11 @@ class CreateAddressForm(BaseForm): address_country = create_address_field("Land", type="country") @property - def address_kwargs(self): + def address_kwargs(self) -> dict[str, str]: return {key: getattr(self, f'address_{key}').data for key in 'street number addition zip_code city state country'.split()} - def set_address_fields(self, obj: RoomAddressSuggestion | Address | None): + def set_address_fields(self, obj: RoomAddressSuggestion | Address | None) -> None: if not obj: return self.address_street.data = obj.street @@ -75,7 +78,7 @@ def set_address_fields(self, obj: RoomAddressSuggestion | Address | None): self.address_addition.data = addition -def building_query(): +def building_query() -> list[Building]: return sort_buildings(Building.q.all()) diff --git a/web/blueprints/helpers/form.py b/web/blueprints/helpers/form.py index 66cb9ba67..077a366c0 100644 --- a/web/blueprints/helpers/form.py +++ b/web/blueprints/helpers/form.py @@ -54,8 +54,15 @@ class ConfirmCheckboxField(BooleanField): See `confirmable-error.ts` """ - def __init__(self, label=None, validators=None, false_values=None, **kwargs): - kwargs.setdefault('render_kw', {}) - kwargs.setdefault('default', False) - kwargs['render_kw'].setdefault('data-role', 'confirm-checkbox') + + def __init__( + self, + label: str | None = None, + validators: t.Iterable[t.Any] | None = None, + false_values: t.Iterable[t.Any] | None = None, + **kwargs: t.Any, + ) -> None: + kwargs.setdefault("render_kw", {}) + kwargs.setdefault("default", False) + kwargs["render_kw"].setdefault("data-role", "confirm-checkbox") super().__init__(label, validators, false_values, **kwargs) diff --git a/web/blueprints/helpers/host.py b/web/blueprints/helpers/host.py index 83834f714..56a8b0f25 100644 --- a/web/blueprints/helpers/host.py +++ b/web/blueprints/helpers/host.py @@ -23,7 +23,7 @@ def __init__(self, annex_field: str | None = 'annex'): def annex_set(self, form: Form) -> bool: return bool(self.annex_field and getattr(form, self.annex_field).data) - def __call__(self, form: Form, field: Field): + def __call__(self, form: Form, field: Field) -> None: if not re.match(mac_regex, field.data): return if self.annex_set(form): diff --git a/web/blueprints/helpers/log.py b/web/blueprints/helpers/log.py index fb0d9b401..ad7059750 100644 --- a/web/blueprints/helpers/log.py +++ b/web/blueprints/helpers/log.py @@ -6,6 +6,7 @@ :class:`LogTableExtended`. """ +import typing as t from datetime import datetime, timezone from functools import partial @@ -41,14 +42,22 @@ def format_log_entry(entry: LogEntry, log_type: LogType) -> LogTableRow: format_room_log_entry = partial(format_log_entry, log_type='room') -def radius_description(interface, entry): +class SupportsFormat(t.Protocol): + def __str__(self) -> str: + ... + + def __format__(self, format_spec: t.Any) -> str: + ... + + +def radius_description(interface: SupportsFormat, entry: RadiusLogEntry) -> str: """Build a readable log message from a radius log entry. :param interface: Something whose string representation can be used to inform about the port the radius log happened. For instance, this can be a :class:`SwitchPort` object formatting itself to `switch-wu5-00 (D15)` or similar. - :param RadiusLogEntry entry: A :class:`RadiusLogEntry` as + :param entry: A :class:`RadiusLogEntry` as obtained from a :class:`HadesLogs` lookup. """ prefix = f"{interface} – {entry.mac} – " @@ -61,7 +70,9 @@ def radius_description(interface, entry): return prefix + msg -def format_hades_log_entry(interface: str, entry: RadiusLogEntry) -> LogTableRow: +def format_hades_log_entry( + interface: SupportsFormat, entry: RadiusLogEntry +) -> LogTableRow: """Turn Radius Log entry information into a canonical form This utilizes :py:func:`radius_description` but returns a dict in diff --git a/web/blueprints/infrastructure/forms.py b/web/blueprints/infrastructure/forms.py index 461579875..1d7397ba3 100644 --- a/web/blueprints/infrastructure/forms.py +++ b/web/blueprints/infrastructure/forms.py @@ -1,4 +1,5 @@ from flask_wtf import FlaskForm as Form +from sqlalchemy.orm import Query from wtforms.validators import DataRequired, IPAddress from pycroft.model.net import VLAN @@ -8,7 +9,7 @@ from wtforms_widgets.fields.filters import to_uppercase, to_lowercase -def vlan_query(): +def vlan_query() -> Query: return VLAN.q.order_by(VLAN.vid) From 717649bd854c2ddc307e62aaa5d5e7a4cedccbc0 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:27:54 +0200 Subject: [PATCH 15/18] [web][typing] add type hints for `properties` blueprint --- web/blueprints/properties/__init__.py | 10 +++++----- web/blueprints/properties/forms.py | 5 +++-- 2 files changed, 8 insertions(+), 7 deletions(-) diff --git a/web/blueprints/properties/__init__.py b/web/blueprints/properties/__init__.py index 9a59de86c..d400432e1 100644 --- a/web/blueprints/properties/__init__.py +++ b/web/blueprints/properties/__init__.py @@ -73,7 +73,7 @@ def property_group_create() -> ResponseValue: @bp.route('/property_group//grant/') @access.require('groups_change') -def property_group_grant_property(group_id, property_name) -> ResponseValue: +def property_group_grant_property(group_id: int, property_name: str) -> ResponseValue: property_group = session.session.get(PropertyGroup, group_id) if property_group is None: @@ -89,7 +89,7 @@ def property_group_grant_property(group_id, property_name) -> ResponseValue: @bp.route('/property_group//deny/') @access.require('groups_change') -def property_group_deny_property(group_id, property_name) -> ResponseValue: +def property_group_deny_property(group_id: int, property_name: str) -> ResponseValue: property_group = session.session.get(PropertyGroup, group_id) if property_group is None: @@ -105,7 +105,7 @@ def property_group_deny_property(group_id, property_name) -> ResponseValue: @bp.route('/property_group//remove/') @access.require('groups_change') -def property_group_remove_property(group_id, property_name) -> ResponseValue: +def property_group_remove_property(group_id: int, property_name: str) -> ResponseValue: group = session.session.get(PropertyGroup, group_id) if group is None: @@ -121,7 +121,7 @@ def property_group_remove_property(group_id, property_name) -> ResponseValue: @bp.route('/property_group//edit', methods=['GET', 'POST']) @access.require('groups_change') -def property_group_edit(group_id) -> ResponseValue: +def property_group_edit(group_id: int) -> ResponseValue: group = session.session.get(PropertyGroup, group_id) if group is None: @@ -153,7 +153,7 @@ def property_group_edit(group_id) -> ResponseValue: @bp.route('/property_group//delete', methods=['GET', 'POST']) @access.require('groups_change') -def property_group_delete(group_id) -> ResponseValue: +def property_group_delete(group_id: int) -> ResponseValue: group = session.session.get(PropertyGroup, group_id) if group is None: diff --git a/web/blueprints/properties/forms.py b/web/blueprints/properties/forms.py index 320c0d60a..1e055b942 100644 --- a/web/blueprints/properties/forms.py +++ b/web/blueprints/properties/forms.py @@ -3,6 +3,7 @@ # the Apache License, Version 2.0. See the LICENSE file for details. from flask_wtf import FlaskForm as Form +from sqlalchemy.orm import Query from wtforms.validators import DataRequired, Regexp from pycroft import config @@ -10,11 +11,11 @@ from wtforms_widgets.fields.core import TextField, IntegerField -def property_group_query(): +def property_group_query() -> Query: return PropertyGroup.q.order_by(PropertyGroup.id) -def property_group_user_create_query(): +def property_group_user_create_query() -> Query: return PropertyGroup.q.filter(PropertyGroup.id.in_([ config.member_group_id, config.external_group_id, From f03941d1829db497ad5f24eb4dccc71bfcb21c90 Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:28:07 +0200 Subject: [PATCH 16/18] [web][typing] add type hints in `task` blueprint --- web/blueprints/task/__init__.py | 13 ++++++++----- 1 file changed, 8 insertions(+), 5 deletions(-) diff --git a/web/blueprints/task/__init__.py b/web/blueprints/task/__init__.py index 7010011df..74c2bf442 100644 --- a/web/blueprints/task/__init__.py +++ b/web/blueprints/task/__init__.py @@ -32,7 +32,10 @@ nav = BlueprintNavigation(bp, "Tasks", icon='fa-tasks', blueprint_access=access) -def format_parameters(parameters): +T = t.TypeVar("T", bound=t.MutableMapping[str, t.Any]) + + +def format_parameters(parameters: T) -> T: """Make task parameters human readable by looking up objects behind ids""" # Replace building_id by the buildings short name @@ -100,7 +103,7 @@ def task_row(task: UserTask) -> TaskRow: @bp.route("/user//json") -def json_tasks_for_user(user_id) -> ResponseValue: +def json_tasks_for_user(user_id: int) -> ResponseValue: user = get_user_or_404(user_id) return TableResponse[TaskRow]( items=[task_row(t.cast(UserTask, task)) for task in user.tasks] @@ -128,7 +131,7 @@ def json_user_tasks() -> ResponseValue: ).model_dump() -def get_task_or_404(task_id) -> Task | NoReturn: +def get_task_or_404(task_id: int) -> Task | NoReturn: if task := session.session.get(Task, task_id): return task abort(404) @@ -158,7 +161,7 @@ def manually_execute_user_task(task_id: int) -> ResponseValue: @bp.route("//cancel") @access.require('user_change') -def cancel_user_task(task_id) -> ResponseValue: +def cancel_user_task(task_id: int) -> ResponseValue: task = get_task_or_404(task_id) cancel_task(task, current_user) @@ -170,7 +173,7 @@ def cancel_user_task(task_id) -> ResponseValue: @bp.route("//reschedule", methods=['GET', 'POST']) @access.require('user_change') -def reschedule_user_task(task_id) -> ResponseValue: +def reschedule_user_task(task_id: int) -> ResponseValue: task = get_task_or_404(task_id) form = RescheduleTaskForm() From 9d6e43bc4ec4fa6e954de1aaf6d72545dc59e0dc Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 12:28:21 +0200 Subject: [PATCH 17/18] [web][typing] fix last remainign mypy admonitions --- hades_logs/__init__.py | 5 ++++- web/form/widgets.py | 8 ++++---- 2 files changed, 8 insertions(+), 5 deletions(-) diff --git a/hades_logs/__init__.py b/hades_logs/__init__.py index 8b353da19..408c52fcc 100644 --- a/hades_logs/__init__.py +++ b/hades_logs/__init__.py @@ -5,6 +5,7 @@ This module provides access to Hades' radius logs utilizing its celery RPC api. """ +import typing as t import logging from celery.exceptions import TimeoutError as CeleryTimeoutError @@ -110,7 +111,9 @@ def create_task(self, name, *args, **kwargs): full_task_name = f'{self.celery.main}.{name}' return self.celery.signature(full_task_name, args=args, kwargs=kwargs) - def fetch_logs(self, nasipaddress: str, nasportid: str, limit=100, reduced=True): + def fetch_logs( + self, nasipaddress: str, nasportid: str, limit: int = 100, reduced: bool = True + ) -> t.Iterator[RadiusLogEntry]: """Fetch the auth logs of the given port :param ipaddr nasipaddress: The IP address of the NAS diff --git a/web/form/widgets.py b/web/form/widgets.py index 97168e0e6..ad2b765c5 100644 --- a/web/form/widgets.py +++ b/web/form/widgets.py @@ -1,5 +1,5 @@ import wtforms_widgets -from wtforms import ValidationError +from wtforms import ValidationError, Form from pycroft.model import session from pycroft.model.user import User @@ -8,15 +8,15 @@ class UserIDField(wtforms_widgets.fields.core.StringField): """A User-ID Field """ - def __init__(self, *args, **kwargs): + def __init__(self, *args, **kwargs) -> None: super().__init__(*args, **kwargs) - def __call__(self, **kwargs): + def __call__(self, **kwargs) -> None: return super().__call__( **kwargs ) - def pre_validate(self, form): + def pre_validate(self, form: Form) -> None: try: id = int(self.data) except ValueError: From 6a5ae156e6d81b3c134e81685a6b6a3197d895dd Mon Sep 17 00:00:00 2001 From: Lukas Juhrich Date: Thu, 24 Aug 2023 13:11:14 +0200 Subject: [PATCH 18/18] [web][typing] fix table tests by using keyword-only arguments --- tests/table/test_bootstrap_table.py | 12 +++++++++--- 1 file changed, 9 insertions(+), 3 deletions(-) diff --git a/tests/table/test_bootstrap_table.py b/tests/table/test_bootstrap_table.py index 9e36e5349..bfa259620 100644 --- a/tests/table/test_bootstrap_table.py +++ b/tests/table/test_bootstrap_table.py @@ -205,7 +205,7 @@ def test_table_args_set(self, EmptyTable): } def test_table_args_after_instantiation(self, EmptyTable): - assert EmptyTable("#").table_args == { + assert EmptyTable(data_url="#").table_args == { 'data-toggle': "table", "data-icons-prefix": "fa", 'data-url': "#", @@ -250,10 +250,16 @@ class Meta: return A_ def test_url_param_is_added(self, A: type[BootstrapTable]): - assert A("http://localhost/table").data_url == "http://localhost/table?inverted=yes" + assert ( + A(data_url="http://localhost/table").data_url + == "http://localhost/table?inverted=yes" + ) def test_url_param_is_overridden(self, A: type[BootstrapTable]): - assert A("http://localhost/table?inverted=no").data_url == "http://localhost/table?inverted=yes" + assert ( + A(data_url="http://localhost/table?inverted=no").data_url + == "http://localhost/table?inverted=yes" + ) class TestSplittedTable: