Resolve nits (#932)

* Add type hints to principal

* Add type hints for database access

* Correct type hints for command line options

* Fix unauthenticated server state

* Fix 3.9 TypeHints

Author: DiamondJoseph

CHANGELOG.md

@@ -9,6 +9,10 @@ Write the date in place of the "Unreleased" in the case a new version is release
 
 - `Composite` structure family to enable direct access to table columns in a single namespace.
 
+### Changed
+
+- Adjust arguments of `print_admin_api_key_if_generated` and rename `print_server_info`
+
 ### Maintenance
 
 - Extract API key handling
@@ -18,7 +22,7 @@ Write the date in place of the "Unreleased" in the case a new version is release
   - This is a breaking change if setting TILED_SERVER_SECRET_KEYS or TILED_ALLOW_ORIGINS,
     TILED_SERVER_SECRET_KEYS is now TILED_SECRET_KEYS and these fields now require passing a json
     list e.g. ``TILED_SECRET_KEYS='["one", "two"]'``
-
+- More type hinting
 
 ## 0.1.0-b20 (2025-03-07)
 

tiled/authn_database/connection_pool.py

@@ -1,16 +1,19 @@
+from collections.abc import AsyncGenerator
+from typing import Optional
+
 from fastapi import Depends
-from sqlalchemy.ext.asyncio import AsyncSession, create_async_engine
+from sqlalchemy.ext.asyncio import AsyncEngine, AsyncSession, create_async_engine
 
-from ..server.settings import Settings, get_settings
+from ..server.settings import DatabaseSettings, Settings, get_settings
 from ..utils import ensure_specified_sql_driver
 
 # A given process probably only has one of these at a time, but we
 # key on database_settings just case in some testing context or something
 # we have two servers running in the same process.
-_connection_pools = {}
+_connection_pools: dict[DatabaseSettings, AsyncEngine] = {}
 
 
-def open_database_connection_pool(database_settings):
+def open_database_connection_pool(database_settings: DatabaseSettings) -> AsyncEngine:
     connect_args = {}
     kwargs = {}  # extra kwargs passed to create_engine
     # kwargs["pool_size"] = database_settings.pool_size
@@ -25,13 +28,15 @@ def open_database_connection_pool(database_settings):
     return engine
 
 
-async def close_database_connection_pool(database_settings):
+async def close_database_connection_pool(database_settings: DatabaseSettings):
     engine = _connection_pools.pop(database_settings, None)
     if engine is not None:
         await engine.dispose()
 
 
-async def get_database_engine(settings: Settings = Depends(get_settings)):
+async def get_database_engine(
+    settings: Settings = Depends(get_settings),
+) -> AsyncEngine:
     # Special case for single-user mode
     if settings.database_settings.uri is None:
         return None
@@ -43,7 +48,9 @@ async def get_database_engine(settings: Settings = Depends(get_settings)):
         )
 
 
-async def get_database_session(engine=Depends(get_database_engine)):
+async def get_database_session(
+    engine: AsyncEngine = Depends(get_database_engine),
+) -> AsyncGenerator[Optional[AsyncSession]]:
     # Special case for single-user mode
     if engine is None:
         yield None

tiled/commandline/_serve.py

@@ -36,7 +36,7 @@ def serve_directory(
             "this option selected."
         ),
     ),
-    api_key: str = typer.Option(
+    api_key: Optional[str] = typer.Option(
         None,
         "--api-key",
         help=(
@@ -55,7 +55,7 @@ def serve_directory(
             "may break user (client-side) code."
         ),
     ),
-    ext: List[str] = typer.Option(
+    ext: Optional[List[str]] = typer.Option(
         None,
         "--ext",
         help=(
@@ -64,7 +64,7 @@ def serve_directory(
             "extension."
         ),
     ),
-    mimetype_detection_hook: str = typer.Option(
+    mimetype_detection_hook: Optional[str] = typer.Option(
         None,
         "--mimetype-hook",
         help=(
@@ -73,15 +73,15 @@ def serve_directory(
             "Specify here as 'package.module:function'"
         ),
     ),
-    adapters: List[str] = typer.Option(
+    adapters: Optional[List[str]] = typer.Option(
         None,
         "--adapter",
         help=(
             "ADVANCED: Custom Tiled Adapter for reading a given format"
             "Specify here as 'mimetype=package.module:function'"
         ),
     ),
-    walkers: List[str] = typer.Option(
+    walkers: Optional[List[str]] = typer.Option(
         None,
         "--walker",
         help=(
@@ -138,7 +138,7 @@ def serve_directory(
     stamp_head(ALEMBIC_INI_TEMPLATE_PATH, ALEMBIC_DIR, database)
 
     from ..catalog import from_uri as catalog_from_uri
-    from ..server.app import build_app, print_admin_api_key_if_generated
+    from ..server.app import build_app, print_server_info
 
     server_settings = {}
     if keep_ext:
@@ -212,7 +212,7 @@ def serve_directory(
 
     from ..client import from_uri as client_from_uri
 
-    print_admin_api_key_if_generated(web_app, host=host, port=port, force=generated)
+    print_server_info(web_app, host=host, port=port, include_api_key=generated)
     log_config = _setup_log_config(log_config, log_timestamps)
     config = uvicorn.Config(web_app, host=host, port=port, log_config=log_config)
     server = uvicorn.Server(config)
@@ -287,16 +287,16 @@ async def serve_and_walk():
 
 
 def serve_catalog(
-    database: str = typer.Argument(
+    database: Optional[str] = typer.Argument(
         None, help="A filepath or database URI, e.g. 'catalog.db'"
     ),
-    read: List[str] = typer.Option(
+    read: Optional[List[str]] = typer.Option(
         None,
         "--read",
         "-r",
         help="Locations that the server may read from",
     ),
-    write: List[str] = typer.Option(
+    write: Optional[List[str]] = typer.Option(
         None,
         "--write",
         "-w",
@@ -321,7 +321,7 @@ def serve_catalog(
             "this option selected."
         ),
     ),
-    api_key: str = typer.Option(
+    api_key: Optional[str] = typer.Option(
         None,
         "--api-key",
         help=(
@@ -363,7 +363,7 @@ def serve_catalog(
 
     from ..catalog import from_uri
     from ..catalog.utils import classify_writable_storage
-    from ..server.app import build_app, print_admin_api_key_if_generated
+    from ..server.app import build_app, print_server_info
 
     parsed_database = urllib.parse.urlparse(database)
     if parsed_database.scheme in ("", "file"):
@@ -477,7 +477,9 @@ def serve_catalog(
         server_settings,
         scalable=scalable,
     )
-    print_admin_api_key_if_generated(web_app, host=host, port=port)
+    print_server_info(
+        web_app, host=host, port=port, include_api_key=api_key is not None
+    )
 
     import uvicorn
 
@@ -502,7 +504,7 @@ def serve_pyobject(
             "option selected."
         ),
     ),
-    api_key: str = typer.Option(
+    api_key: Optional[str] = typer.Option(
         None,
         "--api-key",
         help=(
@@ -534,7 +536,7 @@ def serve_pyobject(
     ),
 ):
     "Serve a Tree instance from a Python module."
-    from ..server.app import build_app, print_admin_api_key_if_generated
+    from ..server.app import build_app, print_server_info
     from ..utils import import_object
 
     tree = import_object(object_path)
@@ -548,7 +550,7 @@ def serve_pyobject(
         server_settings,
         scalable=scalable,
     )
-    print_admin_api_key_if_generated(web_app, host=host, port=port)
+    print_server_info(web_app, host=host, port=port, include_api_key=api_key is None)
 
     import uvicorn
 
@@ -569,13 +571,13 @@ def serve_demo(
     port: int = typer.Option(8000, help="Bind to a socket with this port."),
 ):
     "Start a public server with example data."
-    from ..server.app import build_app, print_admin_api_key_if_generated
+    from ..server.app import build_app, print_server_info
     from ..utils import import_object
 
     EXAMPLE = "tiled.examples.generated:tree"
     tree = import_object(EXAMPLE)
     web_app = build_app(tree, {"allow_anonymous_access": True}, {})
-    print_admin_api_key_if_generated(web_app, host=host, port=port)
+    print_server_info(web_app, host=host, port=port, include_api_key=True)
 
     import uvicorn
 
@@ -584,7 +586,7 @@ def serve_demo(
 
 @serve_app.command("config")
 def serve_config(
-    config_path: Path = typer.Argument(
+    config_path: Optional[Path] = typer.Argument(
         None,
         help=(
             "Path to a config file or directory of config files. "
@@ -601,23 +603,23 @@ def serve_config(
             "option selected."
         ),
     ),
-    api_key: str = typer.Option(
+    api_key: Optional[str] = typer.Option(
         None,
         "--api-key",
         help=(
             "Set the single-user API key. "
             "By default, a random key is generated at startup and printed."
         ),
     ),
-    host: str = typer.Option(
+    host: Optional[str] = typer.Option(
         None,
         help=(
             "Bind socket to this host. Use `--host 0.0.0.0` to make the application "
             "available on your local network. IPv6 addresses are supported, for "
             "example: --host `'::'`. Uses value in config by default."
         ),
     ),
-    port: int = typer.Option(
+    port: Optional[int] = typer.Option(
         None, help="Bind to a socket with this port. Uses value in config by default."
     ),
     scalable: bool = typer.Option(
@@ -659,11 +661,7 @@ def serve_config(
 
     # Delay this import so that we can fail faster if config-parsing fails above.
 
-    from ..server.app import (
-        build_app_from_config,
-        logger,
-        print_admin_api_key_if_generated,
-    )
+    from ..server.app import build_app_from_config, logger, print_server_info
 
     # Extract config for uvicorn.
     uvicorn_kwargs = parsed_config.pop("uvicorn", {})
@@ -684,8 +682,11 @@ def serve_config(
     web_app = build_app_from_config(
         parsed_config, source_filepath=config_path, scalable=scalable
     )
-    print_admin_api_key_if_generated(
-        web_app, host=uvicorn_kwargs["host"], port=uvicorn_kwargs["port"]
+    print_server_info(
+        web_app,
+        host=uvicorn_kwargs["host"],
+        port=uvicorn_kwargs["port"],
+        include_api_key=api_key is None,
     )
 
     # Likewise, delay this import.

tiled/server/app.py

@@ -428,6 +428,9 @@ async def unhandled_exception_handler(
                 )
         # And add this authentication_router itself to the app.
         app.include_router(authentication_router, prefix="/api/v1/auth")
+        app.state.authenticated = True
+    else:
+        app.state.authenticated = False
 
     @cache
     def override_get_authenticators():
@@ -450,8 +453,6 @@ def override_get_settings():
         ]:
             if authentication.get(item) is not None:
                 setattr(settings, item, authentication[item])
-        if authentication.get("single_user_api_key") is not None:
-            settings.single_user_api_key_generated = False
         for item in [
             "allow_origins",
             "response_bytesize_limit",
@@ -884,7 +885,7 @@ def app_factory():
     kwargs = construct_build_app_kwargs(parsed_config, source_filepath=config_path)
     web_app = build_app(**kwargs)
     uvicorn_config = parsed_config.get("uvicorn", {})
-    print_admin_api_key_if_generated(
+    print_server_info(
         web_app, host=uvicorn_config.get("host"), port=uvicorn_config.get("port")
     )
     return web_app
@@ -902,16 +903,14 @@ def __getattr__(name):
     raise AttributeError(name)
 
 
-def print_admin_api_key_if_generated(
-    web_app: FastAPI, host: str, port: int, force: bool = False
+def print_server_info(
+    web_app: FastAPI,
+    host: str = "127.0.0.1",
+    port: int = 8000,
+    include_api_key: bool = False,
 ):
-    "Print message to stderr with API key if server-generated (or force=True)."
-    host = host or "127.0.0.1"
-    port = port or 8000
-    settings = web_app.dependency_overrides.get(get_settings, get_settings)()
-    authenticators = web_app.dependency_overrides.get(
-        get_authenticators, get_authenticators
-    )()
+    settings = get_settings()
+
     if settings.allow_anonymous_access:
         print(
             """
@@ -921,7 +920,7 @@ def print_admin_api_key_if_generated(
 """,
             file=sys.stderr,
         )
-    if (not authenticators) and (force or settings.single_user_api_key_generated):
+    if not web_app.state.authenticated and include_api_key:
         print(
             f"""
     Navigate a web browser or connect a Tiled client to: