From 711e4b42eb28ee6def319e1965511a1aad25eb17 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Hubert=20Bere=C5=9B?= Date: Fri, 4 Apr 2025 13:08:48 +0100 Subject: [PATCH 1/5] Clear type errors --- piccolo_admin/endpoints.py | 6 +++--- 1 file changed, 3 insertions(+), 3 deletions(-) diff --git a/piccolo_admin/endpoints.py b/piccolo_admin/endpoints.py index 6405a070..c9baf676 100644 --- a/piccolo_admin/endpoints.py +++ b/piccolo_admin/endpoints.py @@ -49,9 +49,9 @@ from piccolo_api.session_auth.middleware import SessionsAuthBackend from piccolo_api.session_auth.tables import SessionsBase from pydantic import BaseModel, Field, ValidationError +from starlette.exceptions import HTTPException from starlette.middleware import Middleware from starlette.middleware.authentication import AuthenticationMiddleware -from starlette.middleware.exceptions import HTTPException from starlette.requests import Request from starlette.responses import HTMLResponse, JSONResponse, Response from starlette.staticfiles import StaticFiles @@ -411,7 +411,7 @@ class FormConfigResponseModel(BaseModel): description: t.Optional[str] = None -def handle_auth_exception(request: Request, exc: Exception): +def handle_auth_exception(request, exc: Exception): return JSONResponse({"error": "Auth failed"}, status_code=401) @@ -875,7 +875,7 @@ def _get_media_storage( detail="No such column found.", ) - media_storage = media_columns.get(column) + media_storage = media_columns.get(column) # type: ignore if not media_storage: raise HTTPException( From 2d2f8a2ff4bd911e3258116d7121e1e2384f9939 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Hubert=20Bere=C5=9B?= Date: Fri, 4 Apr 2025 13:57:28 +0100 Subject: [PATCH 2/5] Refactor AdminRouter.__init__ --- piccolo_admin/endpoints.py | 475 +++++++++++++++++++++---------------- 1 file changed, 267 insertions(+), 208 deletions(-) diff --git a/piccolo_admin/endpoints.py b/piccolo_admin/endpoints.py index c9baf676..95b364d5 100644 --- a/piccolo_admin/endpoints.py +++ b/piccolo_admin/endpoints.py @@ -411,6 +411,12 @@ class FormConfigResponseModel(BaseModel): description: t.Optional[str] = None +@dataclass +class SessionExpiryConfig: + session_expiry: timedelta + max_session_expiry: timedelta + increase_expiry: t.Optional[timedelta] + def handle_auth_exception(request, exc: Exception): return JSONResponse({"error": "Auth failed"}, status_code=401) @@ -437,14 +443,6 @@ async def log_error(request: Request, exc: HTTPException): class AdminRouter(FastAPI): - """ - The root returns a single page app. The other URLs are REST endpoints. - """ - - table: t.List[Table] = [] - auth_table: t.Type[BaseUser] = BaseUser - template: str = "" - def __init__( self, *tables: t.Union[t.Type[Table], TableConfig], @@ -466,6 +464,64 @@ def __init__( sidebar_links: t.Dict[str, str] = {}, mfa_providers: t.Optional[t.Sequence[MFAProvider]] = None, ) -> None: + self._init_fastapi_app(site_name, allowed_hosts, debug) + + ####################################################################### + # Convert any table arguments which are plain ``Table`` classes into + # ``TableConfig`` instances. + + self.table_configs = self._init_table_configs(tables) + self.table_config_map = { + table_config.table_class._meta.tablename: table_config + for table_config in self.table_configs + } + + self._check_media_storage() + + ####################################################################### + + self.default_language_code = default_language_code + self.translations_map = { + translation.language_code.lower(): translation + for translation in (translations or TRANSLATIONS) + } + + ####################################################################### + + # Public attribute retained for backwards compatibility. To be deprecated + self.auth_table = auth_table + self.site_name = site_name + self.forms = forms + self.read_only = read_only + self.page_size = page_size + self.sidebar_links = sidebar_links + self.form_config_map = {form.slug: form for form in self.forms} + + self.template = self._init_index_template() + + # Private attributes used internally to initialise the FastAPI app. + # Details and implementation subject to change + self._auth_table = auth_table + self._session_table = session_table + self._rate_limit_provider: RateLimitProvider = ( + rate_limit_provider or InMemoryLimitProvider(limit=20, timespan=300) + ) + self._session_expiry_config = SessionExpiryConfig( + session_expiry=session_expiry, + max_session_expiry=max_session_expiry, + increase_expiry=increase_expiry, + ) + self._mfa_providers: t.Sequence[MFAProvider] = mfa_providers or [] + self._production = production + + self._init_app() + + def _init_fastapi_app(self, site_name, allowed_hosts, debug): + """Call the parent FastAPI constructor, which provisions the top-level ASGI app + and ensures .mount etc. are safe to call from our code. + + This implementation adds middleware for CSRF protection and some custom logging. + """ super().__init__( title=site_name, description=f"{site_name} documentation", @@ -482,115 +538,155 @@ def __init__( redoc_url=None, ) - ####################################################################### - # Convert any table arguments which are plain ``Table`` classes into - # ``TableConfig`` instances. + def _init_app(self): + """ + Configure all the Piccolo admin endpoints and sub-apps. The high-level structure of the admin app consists of: + * /api - endpoints protected by auth, providing data for the admin UI and CRUD operations + * /public - endpoints for login/logout etc. that do not require auth and are used on the login page + * /assets - static files for the admin UI + """ + # App for authenticated endpoints: + api_app = self._init_api_app( + superuser_tables=(self._auth_table, self._session_table), + ) - table_configs: t.List[TableConfig] = [] + # Add /user and /change-password endpoints + self._init_api_auth_endpoints(api_app) - for table in tables: - if isinstance(table, TableConfig): - table_configs.append(table) - else: - table_configs.append(TableConfig(table_class=table)) + # Optional components of the API app: + self._init_mfa_provider(api_app) - self.table_configs = sorted( - table_configs, - key=lambda table_config: table_config.table_class._meta.tablename, - ) - self.table_config_map = { - table_config.table_class._meta.tablename: table_config - for table_config in self.table_configs - } + self._init_auth_middleware(api_app) - ####################################################################### - # Make sure columns are configured properly. + # App for unauthenticated endpoints: + public_app = self._init_public_app() - for table_config in table_configs: - table_class = table_config.table_class - for column in table_class._meta.columns: - if column._meta.secret and column._meta.required: - message = ( - f"{table_class._meta.tablename}." - f"{column._meta._name} is using `secret` and " - f"`required` column args which are incompatible. " - f"You may encounter unexpected behavior when using " - f"this table within Piccolo Admin." - ) - colored_warning(message, level=Level.high) + # Add /login and /logout endpoints + self._init_public_auth_endpoints(public_app) + assets_app = StaticFiles(directory=os.path.join(ASSET_PATH, "assets")) ####################################################################### - # Make sure media storage is configured properly. - media_storage = [ - i - for i in itertools.chain( - *[ - table_config.media_storage or [] - for table_config in table_configs - ] - ) - ] + self.add_route(path="/", route=self.get_root, methods=["GET"]) - if len(media_storage) != len(set(media_storage)): - raise ValueError( - "Media storage is misconfigured - multiple columns are saving " - "to the same location." - ) + self.mount(path="/api", app=api_app) + self.mount(path="/public", app=public_app) + self.mount(path="/assets", app=assets_app) - ####################################################################### + def _init_public_app(self): + """Creates a sub-app for public endpoints.""" + public_app = FastAPI( + redoc_url=None, + docs_url=None, + debug=self.debug, + exception_handlers={500: log_error}, + ) + public_app.mount("/docs/", swagger_ui(schema_url="../openapi.json")) - self.default_language_code = default_language_code - self.translations_map = { - translation.language_code.lower(): translation - for translation in (translations or TRANSLATIONS) - } + # We make the meta endpoint available without auth, because it contains + # the site name. + public_app.add_api_route( + "/meta/", + endpoint=self.get_meta, + tags=["Meta"], # type: ignore + ) - ####################################################################### + # The translations are public, because we need them on the login page. + public_app.add_api_route( + "/translations/", + endpoint=self.get_translation_list, # type: ignore + methods=["GET"], + tags=["Translations"], + response_model=TranslationListResponse, + ) - self.auth_table = auth_table - self.site_name = site_name - self.forms = forms - self.read_only = read_only - self.sidebar_links = sidebar_links - self.form_config_map = {form.slug: form for form in self.forms} + public_app.add_api_route( + "/translations/{language_code:str}/", + endpoint=self.get_translation, # type: ignore + methods=["GET"], + tags=["Translations"], + response_model=Translation, + ) - with open(os.path.join(ASSET_PATH, "index.html")) as f: - self.template = f.read() + return public_app - ####################################################################### + def _init_public_auth_endpoints(self, public_app: FastAPI): + """The unauthenticated user is supposed to hit the /login/ endpoint to obtain credentials. + This method modifies a FastAPI app to add /login/ and /logout/ endpoints. + """ + public_app.mount( + path="/login/", + # This rate limiting is to prevent brute forcing password login, + # and MFA codes. + app=RateLimitingMiddleware( + app=session_login( + auth_table=self._auth_table, + session_table=self._session_table, + session_expiry=self._session_expiry_config.session_expiry, + max_session_expiry=self._session_expiry_config.max_session_expiry, + redirect_to=None, + production=self._production, + mfa_providers=self._mfa_providers, + ), + provider=self._rate_limit_provider, + ), + ) + + public_app.add_route( + path="/logout/", + route=session_logout(session_table=self._session_table), # type: ignore + methods=["POST"], + ) + + def _init_auth_middleware(self, api_app: FastAPI): + """Add middleware to a FastAPI to provide auth on all endpoints. + This method produces an instance of starlette's AuthenticationMiddleware and adds it to the app. + """ + auth_middleware = partial( + AuthenticationMiddleware, + backend=SessionsAuthBackend( + auth_table=self._auth_table, + session_table=self._session_table, + admin_only=True, + increase_expiry=self._session_expiry_config.increase_expiry, + ), + on_error=handle_auth_exception, + ) + api_app.add_middleware(auth_middleware) - private_app = FastAPI( + def _init_api_app(self, superuser_tables=tuple()) -> FastAPI: + """Provision the API app with all authenticated endpoints necessary for the admin UI. + This is a FastAPI app, meaning it can have its own middleware and can be mounted on a parent app. + """ + api_app = FastAPI( docs_url=None, redoc_url=None, - debug=debug, + debug=self.debug, exception_handlers={500: log_error}, ) - private_app.mount("/docs/", swagger_ui(schema_url="../openapi.json")) + api_app.mount("/docs/", swagger_ui(schema_url="../openapi.json")) - for table_config in table_configs: + for table_config in self.table_configs: table_class = table_config.table_class visible_column_names = table_config.get_visible_column_names() visible_filter_names = table_config.get_visible_filter_names() - rich_text_columns_names = ( - table_config.get_rich_text_columns_names() - ) + rich_text_columns_names = table_config.get_rich_text_columns_names() media_columns_names = table_config.get_media_columns_names() link_column_name = table_config.get_link_column()._meta.name order_by = table_config.get_order_by() time_resolution = table_config.get_time_resolution() validators = table_config.validators - if table_class in (auth_table, session_table): + if table_class in superuser_tables: validators = validators or Validators() validators.every = [superuser_validators, *validators.every] FastAPIWrapper( root_url=f"/tables/{table_class._meta.tablename}/", - fastapi_app=private_app, + fastapi_app=api_app, piccolo_crud=PiccoloCRUD( table=table_class, - read_only=read_only, - page_size=page_size, + read_only=self.read_only, + page_size=self.page_size, schema_extra={ "visible_column_names": visible_column_names, "visible_filter_names": visible_filter_names, @@ -610,7 +706,7 @@ def __init__( ), ) - private_app.add_api_route( + api_app.add_api_route( path="/tables/", endpoint=self.get_table_list, # type: ignore methods=["GET"], @@ -618,7 +714,7 @@ def __init__( tags=["Tables"], ) - private_app.add_api_route( + api_app.add_api_route( path="/tables/grouped/", endpoint=self.get_table_list_grouped, # type: ignore methods=["GET"], @@ -626,14 +722,14 @@ def __init__( tags=["Tables"], ) - private_app.add_api_route( + api_app.add_api_route( path="/links/", endpoint=self.get_sidebar_links, # type: ignore methods=["GET"], tags=["Links"], ) - private_app.add_api_route( + api_app.add_api_route( path="/forms/", endpoint=self.get_forms, # type: ignore methods=["GET"], @@ -641,7 +737,7 @@ def __init__( response_model=t.List[FormConfigResponseModel], ) - private_app.add_api_route( + api_app.add_api_route( path="/forms/grouped/", endpoint=self.get_grouped_forms, # type: ignore methods=["GET"], @@ -649,28 +745,33 @@ def __init__( tags=["Forms"], ) - private_app.add_api_route( + api_app.add_api_route( path="/forms/{form_slug:str}/", endpoint=self.get_single_form, # type: ignore methods=["GET"], tags=["Forms"], ) - private_app.add_api_route( + api_app.add_api_route( path="/forms/{form_slug:str}/schema/", endpoint=self.get_single_form_schema, # type: ignore methods=["GET"], tags=["Forms"], ) - private_app.add_api_route( + api_app.add_api_route( path="/forms/{form_slug:str}/", endpoint=self.post_single_form, # type: ignore methods=["POST"], tags=["Forms"], ) - private_app.add_api_route( + self._init_media_endpoints(api_app) + return api_app + + def _init_api_auth_endpoints(self, api_app: FastAPI): + """Modify an app by adding endpoints for self-service for an authenticated user: /user/ and /change-password/""" + api_app.add_api_route( path="/user/", endpoint=self.get_user, # type: ignore methods=["GET"], @@ -678,20 +779,19 @@ def __init__( response_model=UserResponseModel, ) - private_app.add_route( + api_app.add_route( path="/change-password/", route=change_password( # type: ignore login_url="./../../public/login/", - session_table=session_table, - read_only=read_only, + session_table=self._session_table, + read_only=self.read_only, ), methods=["POST"], ) - ####################################################################### - # Media - - private_app.add_api_route( + def _init_media_endpoints(self, api_app: FastAPI): + """Add /media/ endpoints required by some functionality of the admin UI.""" + api_app.add_api_route( path="/media/", endpoint=self.store_file, # type: ignore methods=["POST"], @@ -699,7 +799,7 @@ def __init__( response_model=StoreFileResponseModel, ) - private_app.add_api_route( + api_app.add_api_route( path="/media/generate-file-url/", endpoint=self.generate_file_url, # type: ignore methods=["POST"], @@ -709,139 +809,93 @@ def __init__( for table_config in self.table_configs: if table_config.media_columns: - for ( - column, - media_storage, - ) in table_config.media_columns.items(): + for column, media_storage in table_config.media_columns.items(): if isinstance(media_storage, LocalMediaStorage): # We apply a restrictive CSP here to mitigate SVG # files being used maliciously when viewed by admins - private_app.mount( + api_app.mount( path=f"/media-files/{column._meta.table._meta.tablename}/{column._meta.name}/", # noqa: E501 app=CSPMiddleware( - StaticFiles( - directory=media_storage.media_path - ), + StaticFiles(directory=media_storage.media_path), config=CSPConfig(default_src="none"), ), ) - ####################################################################### - # MFA - - if mfa_providers: - if len(mfa_providers) > 1: - raise ValueError( - "Only a single mfa_provider is currently supported." - ) - - for mfa_provider in mfa_providers: - private_app.mount( - path="/mfa-setup/", - # This rate limiting is because some of the forms accept - # a password, and generating recovery codes is somewhat - # expensive, so we want to prevent abuse. - app=RateLimitingMiddleware( - app=mfa_setup( - provider=mfa_provider, - auth_table=self.auth_table, - ), - provider=InMemoryLimitProvider(limit=20, timespan=300), + def _init_mfa_provider(self, api_app): + """Add /mfa-setup/ endpoint to an app.""" + if len(self._mfa_providers) > 1: + raise ValueError("Only a single mfa_provider is currently supported.") + + for mfa_provider in self._mfa_providers: + api_app.mount( + path="/mfa-setup/", + # This rate limiting is because some of the forms accept + # a password, and generating recovery codes is somewhat + # expensive, so we want to prevent abuse. + app=RateLimitingMiddleware( + app=mfa_setup( + provider=mfa_provider, + auth_table=self._auth_table, ), - ) + provider=InMemoryLimitProvider(limit=20, timespan=300), + # Note: this provider is hard-coded; we may consider using self._rate_limit_provider instead + ), + ) - ####################################################################### + def _init_index_template(self): + with open(os.path.join(ASSET_PATH, "index.html")) as f: + return f.read() - public_app = FastAPI( - redoc_url=None, - docs_url=None, - debug=debug, - exception_handlers={500: log_error}, - ) - public_app.mount("/docs/", swagger_ui(schema_url="../openapi.json")) + def _check_media_storage(self): + # Make sure media storage is configured properly. - if not rate_limit_provider: - rate_limit_provider = InMemoryLimitProvider( - limit=20, - timespan=300, + media_storage = [ + i + for i in itertools.chain( + *[ + table_config.media_storage or [] + for table_config in self.table_configs + ] ) + ] - public_app.mount( - path="/login/", - # This rate limiting is to prevent brute forcing password login, - # and MFA codes. - app=RateLimitingMiddleware( - app=session_login( - auth_table=self.auth_table, - session_table=session_table, - session_expiry=session_expiry, - max_session_expiry=max_session_expiry, - redirect_to=None, - production=production, - mfa_providers=mfa_providers, - ), - provider=rate_limit_provider, - ), - ) - - public_app.add_route( - path="/logout/", - route=session_logout(session_table=session_table), # type: ignore - methods=["POST"], - ) - - # We make the meta endpoint available without auth, because it contains - # the site name. - public_app.add_api_route( - "/meta/", endpoint=self.get_meta, tags=["Meta"] # type: ignore - ) + if len(media_storage) != len(set(media_storage)): + raise ValueError( + "Media storage is misconfigured - multiple columns are saving " + "to the same location." + ) - # The translations are public, because we need them on the login page. - public_app.add_api_route( - "/translations/", - endpoint=self.get_translation_list, # type: ignore - methods=["GET"], - tags=["Translations"], - response_model=TranslationListResponse, - ) + def _init_table_configs(self, tables) -> list[TableConfig]: + """Validate and structure information about the tables for the admin UI.""" + table_configs: t.List[TableConfig] = [] - public_app.add_api_route( - "/translations/{language_code:str}/", - endpoint=self.get_translation, # type: ignore - methods=["GET"], - tags=["Translations"], - response_model=Translation, - ) + for table in tables: + if isinstance(table, TableConfig): + table_configs.append(table) + else: + table_configs.append(TableConfig(table_class=table)) ####################################################################### + # Make sure columns are configured properly. - self.router.add_route( - path="/", endpoint=self.get_root, methods=["GET"] - ) - - self.mount( - path="/assets", - app=StaticFiles(directory=os.path.join(ASSET_PATH, "assets")), - ) + for table_config in table_configs: + table_class = table_config.table_class + for column in table_class._meta.columns: + if column._meta.secret and column._meta.required: + message = ( + f"{table_class._meta.tablename}." + f"{column._meta._name} is using `secret` and " + f"`required` column args which are incompatible. " + f"You may encounter unexpected behavior when using " + f"this table within Piccolo Admin." + ) + colored_warning(message, level=Level.high) - auth_middleware = partial( - AuthenticationMiddleware, - backend=SessionsAuthBackend( - auth_table=auth_table, - session_table=session_table, - admin_only=True, - increase_expiry=increase_expiry, - ), - on_error=handle_auth_exception, + return sorted( + table_configs, + key=lambda table_config: table_config.table_class._meta.tablename, ) - self.mount(path="/api", app=auth_middleware(private_app)) - self.mount(path="/public", app=public_app) - - async def get_root(self, request: Request) -> HTMLResponse: - return HTMLResponse(self.template) - - ########################################################################### def _get_media_storage( self, table_name: str, column_name: str @@ -885,6 +939,10 @@ def _get_media_storage( return media_storage + + async def get_root(self, request: Request) -> HTMLResponse: + return HTMLResponse(self.template) + async def store_file( self, request: Request, @@ -1147,6 +1205,7 @@ def get_translation(self, language_code: str = "en") -> Translation: return translation + def get_all_tables( tables: t.Sequence[t.Type[Table]], ) -> t.Sequence[t.Type[Table]]: From b8c61c0d2302f2a6934d97f8b4b00cdbce263c4f Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Hubert=20Bere=C5=9B?= Date: Mon, 7 Apr 2025 08:18:20 +0100 Subject: [PATCH 3/5] Review tweaks: type hints, comments --- piccolo_admin/endpoints.py | 3 +-- 1 file changed, 1 insertion(+), 2 deletions(-) diff --git a/piccolo_admin/endpoints.py b/piccolo_admin/endpoints.py index 95b364d5..9ff83fb3 100644 --- a/piccolo_admin/endpoints.py +++ b/piccolo_admin/endpoints.py @@ -516,7 +516,7 @@ def __init__( self._init_app() - def _init_fastapi_app(self, site_name, allowed_hosts, debug): + def _init_fastapi_app(self, site_name: str, allowed_hosts: t.Sequence[str], debug: bool): """Call the parent FastAPI constructor, which provisions the top-level ASGI app and ensures .mount etc. are safe to call from our code. @@ -838,7 +838,6 @@ def _init_mfa_provider(self, api_app): auth_table=self._auth_table, ), provider=InMemoryLimitProvider(limit=20, timespan=300), - # Note: this provider is hard-coded; we may consider using self._rate_limit_provider instead ), ) From 0a560733ef33fd6ae19e5e06e2cfed2de6f467a0 Mon Sep 17 00:00:00 2001 From: =?UTF-8?q?Hubert=20Bere=C5=9B?= Date: Mon, 7 Apr 2025 08:18:28 +0100 Subject: [PATCH 4/5] Fit within 79 characters --- piccolo_admin/endpoints.py | 66 ++++++++++++++++++++++++-------------- 1 file changed, 42 insertions(+), 24 deletions(-) diff --git a/piccolo_admin/endpoints.py b/piccolo_admin/endpoints.py index 9ff83fb3..85d3c922 100644 --- a/piccolo_admin/endpoints.py +++ b/piccolo_admin/endpoints.py @@ -417,6 +417,7 @@ class SessionExpiryConfig: max_session_expiry: timedelta increase_expiry: t.Optional[timedelta] + def handle_auth_exception(request, exc: Exception): return JSONResponse({"error": "Auth failed"}, status_code=401) @@ -504,7 +505,8 @@ def __init__( self._auth_table = auth_table self._session_table = session_table self._rate_limit_provider: RateLimitProvider = ( - rate_limit_provider or InMemoryLimitProvider(limit=20, timespan=300) + rate_limit_provider + or InMemoryLimitProvider(limit=20, timespan=300) ) self._session_expiry_config = SessionExpiryConfig( session_expiry=session_expiry, @@ -516,11 +518,13 @@ def __init__( self._init_app() - def _init_fastapi_app(self, site_name: str, allowed_hosts: t.Sequence[str], debug: bool): - """Call the parent FastAPI constructor, which provisions the top-level ASGI app - and ensures .mount etc. are safe to call from our code. + def _init_fastapi_app( + self, site_name: str, allowed_hosts: t.Sequence[str], debug: bool + ): + """Call the parent FastAPI constructor, which provisions the top-level + ASGI app and ensures .mount etc. are safe to call from our code. - This implementation adds middleware for CSRF protection and some custom logging. + This adds middleware for CSRF protection and custom logging. """ super().__init__( title=site_name, @@ -540,9 +544,11 @@ def _init_fastapi_app(self, site_name: str, allowed_hosts: t.Sequence[str], debu def _init_app(self): """ - Configure all the Piccolo admin endpoints and sub-apps. The high-level structure of the admin app consists of: - * /api - endpoints protected by auth, providing data for the admin UI and CRUD operations - * /public - endpoints for login/logout etc. that do not require auth and are used on the login page + Configure all the Piccolo admin endpoints and sub-apps. + The high-level structure of the admin app consists of: + * /api - endpoints protected by auth, used for the admin UI + * /public - endpoints for login/logout etc. that do not require auth + and are used on the login page * /assets - static files for the admin UI """ # App for authenticated endpoints: @@ -611,8 +617,9 @@ def _init_public_app(self): return public_app def _init_public_auth_endpoints(self, public_app: FastAPI): - """The unauthenticated user is supposed to hit the /login/ endpoint to obtain credentials. - This method modifies a FastAPI app to add /login/ and /logout/ endpoints. + """The unauthenticated user is supposed to hit the /login/ endpoint to + obtain credentials. + This modifies a FastAPI app to add /login/ and /logout/ endpoints. """ public_app.mount( path="/login/", @@ -640,7 +647,8 @@ def _init_public_auth_endpoints(self, public_app: FastAPI): def _init_auth_middleware(self, api_app: FastAPI): """Add middleware to a FastAPI to provide auth on all endpoints. - This method produces an instance of starlette's AuthenticationMiddleware and adds it to the app. + This produces an instance of starlette's AuthenticationMiddleware and + adds it to the app. """ auth_middleware = partial( AuthenticationMiddleware, @@ -655,8 +663,10 @@ def _init_auth_middleware(self, api_app: FastAPI): api_app.add_middleware(auth_middleware) def _init_api_app(self, superuser_tables=tuple()) -> FastAPI: - """Provision the API app with all authenticated endpoints necessary for the admin UI. - This is a FastAPI app, meaning it can have its own middleware and can be mounted on a parent app. + """Provision the API app with all authenticated endpoints necessary for + the admin UI. + This is a FastAPI app, meaning it can have its own middleware and can + be mounted on a parent app. """ api_app = FastAPI( docs_url=None, @@ -670,7 +680,9 @@ def _init_api_app(self, superuser_tables=tuple()) -> FastAPI: table_class = table_config.table_class visible_column_names = table_config.get_visible_column_names() visible_filter_names = table_config.get_visible_filter_names() - rich_text_columns_names = table_config.get_rich_text_columns_names() + rich_text_columns_names = ( + table_config.get_rich_text_columns_names() + ) media_columns_names = table_config.get_media_columns_names() link_column_name = table_config.get_link_column()._meta.name order_by = table_config.get_order_by() @@ -770,7 +782,9 @@ def _init_api_app(self, superuser_tables=tuple()) -> FastAPI: return api_app def _init_api_auth_endpoints(self, api_app: FastAPI): - """Modify an app by adding endpoints for self-service for an authenticated user: /user/ and /change-password/""" + """Modify an app by adding endpoints for self-service for an + authenticated user: /user/ and /change-password/ + """ api_app.add_api_route( path="/user/", endpoint=self.get_user, # type: ignore @@ -790,7 +804,7 @@ def _init_api_auth_endpoints(self, api_app: FastAPI): ) def _init_media_endpoints(self, api_app: FastAPI): - """Add /media/ endpoints required by some functionality of the admin UI.""" + """Add /media/ endpoints required by some admin UI functionality.""" api_app.add_api_route( path="/media/", endpoint=self.store_file, # type: ignore @@ -809,14 +823,19 @@ def _init_media_endpoints(self, api_app: FastAPI): for table_config in self.table_configs: if table_config.media_columns: - for column, media_storage in table_config.media_columns.items(): + for ( + column, + media_storage, + ) in table_config.media_columns.items(): if isinstance(media_storage, LocalMediaStorage): # We apply a restrictive CSP here to mitigate SVG # files being used maliciously when viewed by admins api_app.mount( path=f"/media-files/{column._meta.table._meta.tablename}/{column._meta.name}/", # noqa: E501 app=CSPMiddleware( - StaticFiles(directory=media_storage.media_path), + StaticFiles( + directory=media_storage.media_path + ), config=CSPConfig(default_src="none"), ), ) @@ -824,7 +843,9 @@ def _init_media_endpoints(self, api_app: FastAPI): def _init_mfa_provider(self, api_app): """Add /mfa-setup/ endpoint to an app.""" if len(self._mfa_providers) > 1: - raise ValueError("Only a single mfa_provider is currently supported.") + raise ValueError( + "Only a single mfa_provider is currently supported." + ) for mfa_provider in self._mfa_providers: api_app.mount( @@ -865,7 +886,7 @@ def _check_media_storage(self): ) def _init_table_configs(self, tables) -> list[TableConfig]: - """Validate and structure information about the tables for the admin UI.""" + """Validate and structure information about the database tables.""" table_configs: t.List[TableConfig] = [] for table in tables: @@ -895,7 +916,6 @@ def _init_table_configs(self, tables) -> list[TableConfig]: key=lambda table_config: table_config.table_class._meta.tablename, ) - def _get_media_storage( self, table_name: str, column_name: str ) -> MediaStorage: @@ -928,7 +948,7 @@ def _get_media_storage( detail="No such column found.", ) - media_storage = media_columns.get(column) # type: ignore + media_storage = media_columns.get(column) # type: ignore if not media_storage: raise HTTPException( @@ -938,7 +958,6 @@ def _get_media_storage( return media_storage - async def get_root(self, request: Request) -> HTMLResponse: return HTMLResponse(self.template) @@ -1204,7 +1223,6 @@ def get_translation(self, language_code: str = "en") -> Translation: return translation - def get_all_tables( tables: t.Sequence[t.Type[Table]], ) -> t.Sequence[t.Type[Table]]: From e125657172838edadc800191927d603dc3a23915 Mon Sep 17 00:00:00 2001 From: hubert-springbok <115166625+hubert-springbok@users.noreply.github.com> Date: Mon, 7 Apr 2025 19:38:25 +0100 Subject: [PATCH 5/5] Add type hints Apply suggestions from code review Co-authored-by: Ethan <47520067+Skelmis@users.noreply.github.com> --- piccolo_admin/endpoints.py | 12 ++++++------ 1 file changed, 6 insertions(+), 6 deletions(-) diff --git a/piccolo_admin/endpoints.py b/piccolo_admin/endpoints.py index 85d3c922..838506d6 100644 --- a/piccolo_admin/endpoints.py +++ b/piccolo_admin/endpoints.py @@ -415,7 +415,7 @@ class FormConfigResponseModel(BaseModel): class SessionExpiryConfig: session_expiry: timedelta max_session_expiry: timedelta - increase_expiry: t.Optional[timedelta] + increase_expiry: t.Optional[timedelta] = timedelta(minutes=20) def handle_auth_exception(request, exc: Exception): @@ -579,7 +579,7 @@ def _init_app(self): self.mount(path="/public", app=public_app) self.mount(path="/assets", app=assets_app) - def _init_public_app(self): + def _init_public_app(self) -> FastAPI: """Creates a sub-app for public endpoints.""" public_app = FastAPI( redoc_url=None, @@ -662,7 +662,7 @@ def _init_auth_middleware(self, api_app: FastAPI): ) api_app.add_middleware(auth_middleware) - def _init_api_app(self, superuser_tables=tuple()) -> FastAPI: + def _init_api_app(self, superuser_tables: t.Tuple[t.Type[Table]] = tuple()) -> FastAPI: """Provision the API app with all authenticated endpoints necessary for the admin UI. This is a FastAPI app, meaning it can have its own middleware and can @@ -840,7 +840,7 @@ def _init_media_endpoints(self, api_app: FastAPI): ), ) - def _init_mfa_provider(self, api_app): + def _init_mfa_provider(self, api_app: FastAPI): """Add /mfa-setup/ endpoint to an app.""" if len(self._mfa_providers) > 1: raise ValueError( @@ -862,7 +862,7 @@ def _init_mfa_provider(self, api_app): ), ) - def _init_index_template(self): + def _init_index_template(self) -> str: with open(os.path.join(ASSET_PATH, "index.html")) as f: return f.read() @@ -885,7 +885,7 @@ def _check_media_storage(self): "to the same location." ) - def _init_table_configs(self, tables) -> list[TableConfig]: + def _init_table_configs(self, tables: t.Sequence[t.Union[t.Type[Table], TableConfig]]) -> t.List[TableConfig]: """Validate and structure information about the database tables.""" table_configs: t.List[TableConfig] = []