Skip to content

Commit

Permalink
[typing] add type hints to web.app
Browse files Browse the repository at this point in the history
  • Loading branch information
lukasjuhrich committed Aug 27, 2023
1 parent f6196bb commit 4063fc8
Show file tree
Hide file tree
Showing 2 changed files with 34 additions and 25 deletions.
5 changes: 4 additions & 1 deletion stubs/flask_login/login_manager.pyi
Original file line number Diff line number Diff line change
@@ -1,4 +1,7 @@
import typing as t

from flask.typing import ResponseReturnValue

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
Expand All @@ -22,7 +25,7 @@ class LoginManager:
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): ...
def unauthorized(self) -> ResponseReturnValue: ...
_C = t.TypeVar("_C")
def user_loader(self, callback: _C) -> _C: ...
@property
Expand Down
54 changes: 30 additions & 24 deletions web/app.py
Original file line number Diff line number Diff line change
Expand Up @@ -5,13 +5,19 @@
import logging
import sentry_sdk
from flask import (
Flask, current_app, redirect, render_template, request, url_for,
Flask,
current_app,
redirect,
render_template,
request,
url_for,
g,
make_response,
Response,
)
from flask.typing import ResponseValue
from flask.typing import ResponseValue, ResponseReturnValue
from flask_babel import Babel
from flask_login import current_user
from flask_login import current_user, LoginManager
from jinja2 import select_autoescape
from werkzeug.datastructures import ImmutableDict
from sentry_sdk.integrations.flask import FlaskIntegration
Expand Down Expand Up @@ -47,7 +53,9 @@ class PycroftFlask(Flask):
undefined=jinja2.StrictUndefined,
)

def __init__(self, *a, **kw):
login_manager: LoginManager

def __init__(self, *a: t.Any, **kw: t.Any) -> None:
super().__init__(*a, **kw)
# config keys to support:
self.maybe_add_config_from_env([
Expand All @@ -59,13 +67,11 @@ def __init__(self, *a, **kw):
'HADES_ROUTING_KEY',
])

def maybe_add_config_from_env(self, keys):
def maybe_add_config_from_env(self, keys: t.Iterable[str]) -> None:
"""Write keys from the environment to the app's config
If a key does not exist in the environment, it will just be
skipped.
:param keys: An iterable of strings
"""
for key in keys:
try:
Expand All @@ -77,11 +83,8 @@ def maybe_add_config_from_env(self, keys):
self.logger.debug("Config key %s successfuly read from environment", key)


def make_app(debug=False, hades_logs=True):
""" Create and configure the main? Flask app object
:return: The fully configured app object
"""
def make_app(debug: bool = False, hades_logs: bool = True) -> PycroftFlask:
"""Create and configure the main? Flask app object"""
app = PycroftFlask(__name__)
app.debug = debug

Expand Down Expand Up @@ -122,7 +125,7 @@ def make_app(debug=False, hades_logs=True):
@app.errorhandler(403)
@app.errorhandler(404)
@app.errorhandler(500)
def errorpage(e):
def errorpage(e: Exception) -> ResponseReturnValue:
"""Handle errors according to their error code
:param e: The error from the errorhandler
Expand All @@ -133,10 +136,8 @@ def errorpage(e):
if request.path.startswith('/api/'):
return api.errorpage(e)

if not hasattr(e, 'code'):
code = 500
else:
code = e.code
code = getattr(e, "code", 500)

if code == 500:
message = str(e)
elif code == 403:
Expand All @@ -158,9 +159,10 @@ def debug_sentry() -> t.NoReturn:
app.logger.error("Someone used the debug-sentry endpoint! Also, this is a test error.",
extra={'pi': 3.141})
div_by_zero = 1 / 0 # noqa
assert False # noqa: B011

@app.teardown_request
def shutdown_session(exception=None):
def shutdown_session(exception: BaseException | None = None) -> None:
if app.testing:
# things are not necessarily committed here,
# so `remove` would result in a `ROLLBACK TO SAVEPOINT` to a pre-setup state.
Expand All @@ -169,14 +171,16 @@ def shutdown_session(exception=None):
session.Session.remove()

@app.before_request
def require_login():
def require_login() -> ResponseReturnValue | None:
"""Request a login for every page
except the login blueprint and the static folder.
Blueprint "None" is needed for "/static/*" GET requests.
"""
if current_user.is_anonymous and request.blueprint not in ("login", 'api', None):
return current_app.login_manager.unauthorized()
lm = t.cast(LoginManager, current_app.login_manager) # type: ignore[attr-defined]
return lm.unauthorized()
return None

if app.debug:
register_pyinstrument(app)
Expand All @@ -185,21 +189,21 @@ def require_login():
return app


def register_pyinstrument(app: Flask):
def register_pyinstrument(app: Flask) -> None:
try:
from pyinstrument import Profiler
except ImportError:
app.logger.info("in debug mode, but pyinstrument not installed.")
return

@app.before_request
def before_request():
def before_request() -> None:
if "profile" in request.args:
g.profiler = Profiler()
g.profiler.start()

@app.after_request
def after_request(response):
def after_request(response: Response) -> Response:
if not hasattr(g, "profiler"):
return response
g.profiler.stop()
Expand All @@ -210,7 +214,9 @@ def after_request(response):


if dsn := os.getenv('PYCROFT_SENTRY_DSN'):
def before_send(event, hint):
_TE = t.TypeVar("_TE")

def before_send(event: _TE, hint: dict[str, t.Any]) -> _TE | None:
if 'exc_info' in hint:
exc_type, exc_value, _tb = hint['exc_info']
if isinstance(exc_value, IGNORED_EXCEPTION_TYPES):
Expand Down

0 comments on commit 4063fc8

Please sign in to comment.