From f3fd95d9235194423c0c8a274f4a1e93f383e712 Mon Sep 17 00:00:00 2001 From: Brendon Smith Date: Mon, 5 Jul 2021 15:01:30 -0400 Subject: [PATCH] Update type annotations for Python 3.10 This commit will update type annotation syntax for Python 3.10. The project currently also supports Python 3.8 and 3.9, so the annotations are imported with `from __future__ import annotations`. The Python 3.10 union operator (the pipe, like `str | None`) will not be used on pydantic models. If running Python 3.9 or below, pydantic is not compatible with the union operator, even if annotations are imported with `from __future__ import annotations`. https://peps.python.org/pep-0604/ https://docs.python.org/3/whatsnew/3.10.html https://github.com/samuelcolvin/pydantic/issues/2597#issuecomment-1086194186 https://github.com/samuelcolvin/pydantic/pull/2609#issuecomment-1084341812 https://github.com/samuelcolvin/pydantic/issues/3300#issuecomment-1034007897 --- inboard/app/main_base.py | 12 +++++++----- inboard/app/utilities_starlette.py | 5 +++-- inboard/gunicorn_conf.py | 7 ++++--- inboard/logging_conf.py | 9 +++++---- inboard/start.py | 9 +++++---- tests/app/test_main.py | 9 +++++---- tests/test_gunicorn_conf.py | 7 ++++--- 7 files changed, 33 insertions(+), 25 deletions(-) diff --git a/inboard/app/main_base.py b/inboard/app/main_base.py index 9554522..185913e 100644 --- a/inboard/app/main_base.py +++ b/inboard/app/main_base.py @@ -1,6 +1,8 @@ +from __future__ import annotations + import os import sys -from typing import Awaitable, Callable, Dict +from typing import Awaitable, Callable class App: @@ -9,13 +11,13 @@ class App: https://www.uvicorn.org/ """ - def __init__(self, scope: Dict) -> None: + def __init__(self, scope: dict) -> None: assert scope["type"] == "http" self.scope = scope async def __call__( - self, receive: Dict, send: Callable[[Dict], Awaitable] - ) -> Dict[str, str]: + self, receive: dict, send: Callable[[dict], Awaitable] + ) -> dict[str, str]: await send( { "type": "http.response.start", @@ -32,7 +34,7 @@ async def __call__( raise NameError("Process manager needs to be either uvicorn or gunicorn.") server = "Uvicorn" if process_manager == "uvicorn" else "Uvicorn, Gunicorn," message = f"Hello World, from {server} and Python {version}!" - response: Dict = {"type": "http.response.body", "body": message.encode("utf-8")} + response: dict = {"type": "http.response.body", "body": message.encode("utf-8")} await send(response) return response diff --git a/inboard/app/utilities_starlette.py b/inboard/app/utilities_starlette.py index ef1d427..b125d2d 100644 --- a/inboard/app/utilities_starlette.py +++ b/inboard/app/utilities_starlette.py @@ -1,7 +1,8 @@ +from __future__ import annotations + import base64 import os import secrets -from typing import Optional, Tuple from starlette.authentication import ( AuthCredentials, @@ -17,7 +18,7 @@ class BasicAuth(AuthenticationBackend): async def authenticate( self, request: HTTPConnection - ) -> Optional[Tuple[AuthCredentials, SimpleUser]]: + ) -> tuple[AuthCredentials, SimpleUser] | None: """Authenticate a Starlette request with HTTP Basic auth.""" if "Authorization" not in request.headers: return None diff --git a/inboard/gunicorn_conf.py b/inboard/gunicorn_conf.py index 9e425d6..c701c78 100644 --- a/inboard/gunicorn_conf.py +++ b/inboard/gunicorn_conf.py @@ -1,13 +1,14 @@ +from __future__ import annotations + import multiprocessing import os -from typing import Optional from inboard.logging_conf import configure_logging def calculate_workers( - max_workers: Optional[str] = None, - total_workers: Optional[str] = None, + max_workers: str | None = None, + total_workers: str | None = None, workers_per_core: str = "1", ) -> int: """Calculate the number of Gunicorn worker processes.""" diff --git a/inboard/logging_conf.py b/inboard/logging_conf.py index c9a41b7..8981611 100644 --- a/inboard/logging_conf.py +++ b/inboard/logging_conf.py @@ -1,10 +1,11 @@ +from __future__ import annotations + import importlib.util import logging import logging.config import os import sys from pathlib import Path -from typing import Optional, Set def find_and_load_logging_conf(logging_conf: str) -> dict: @@ -30,7 +31,7 @@ def find_and_load_logging_conf(logging_conf: str) -> dict: def configure_logging( logger: logging.Logger = logging.getLogger(), - logging_conf: Optional[str] = os.getenv("LOGGING_CONF"), + logging_conf: str | None = os.getenv("LOGGING_CONF"), ) -> dict: """Configure Python logging given the name of a logging module or file.""" try: @@ -67,7 +68,7 @@ class LogFilter(logging.Filter): def __init__( self, name: str = "", - filters: Optional[Set[str]] = None, + filters: set[str] | None = None, ) -> None: """Initialize a filter.""" self.name = name @@ -85,7 +86,7 @@ def filter(self, record: logging.LogRecord) -> bool: return all(match not in message for match in self.filters) @staticmethod - def set_filters(input_filters: Optional[str] = None) -> Optional[Set[str]]: + def set_filters(input_filters: str | None = None) -> set[str] | None: """Set log message filters. Filters identify log messages to filter out, so that the logger does not diff --git a/inboard/start.py b/inboard/start.py index 56fab46..eed5ec7 100644 --- a/inboard/start.py +++ b/inboard/start.py @@ -1,11 +1,12 @@ #!/usr/bin/env python3 +from __future__ import annotations + import importlib.util import json import logging import os import subprocess from pathlib import Path -from typing import Optional import uvicorn # type: ignore @@ -52,7 +53,7 @@ def set_gunicorn_options(app_module: str) -> list: return ["gunicorn", "-k", worker_class, "-c", gunicorn_conf_path, app_module] -def _split_uvicorn_option(option: str) -> Optional[list]: +def _split_uvicorn_option(option: str) -> list | None: return ( [option_item.strip() for option_item in str(option_value).split(sep=",")] if (option_value := os.getenv(option.upper())) @@ -77,7 +78,7 @@ def _update_uvicorn_config_options(uvicorn_config_options: dict) -> dict: return uvicorn_config_options -def set_uvicorn_options(log_config: Optional[dict] = None) -> dict: +def set_uvicorn_options(log_config: dict | None = None) -> dict: """Set options for running the Uvicorn server.""" host = os.getenv("HOST", "0.0.0.0") port = int(os.getenv("PORT", "80")) @@ -99,7 +100,7 @@ def start_server( process_manager: str, app_module: str, logger: logging.Logger = logging.getLogger(), - logging_conf_dict: Optional[dict] = None, + logging_conf_dict: dict | None = None, ) -> None: """Start the Uvicorn or Gunicorn server.""" try: diff --git a/tests/app/test_main.py b/tests/app/test_main.py index da47b41..d544ef0 100644 --- a/tests/app/test_main.py +++ b/tests/app/test_main.py @@ -1,5 +1,6 @@ +from __future__ import annotations + import sys -from typing import Dict, List import pytest from fastapi import FastAPI @@ -14,7 +15,7 @@ class TestCors: [Starlette CORS docs](https://www.starlette.io/middleware/#corsmiddleware). """ - origins: Dict[str, List[str]] = { + origins: dict[str, list[str]] = { "allowed": [ "http://br3ndon.land", "https://br3ndon.land", @@ -38,7 +39,7 @@ def test_cors_preflight_response_allowed( self, allowed_origin: str, client: TestClient ) -> None: """Test pre-flight response to cross-origin request from allowed origin.""" - headers: Dict[str, str] = { + headers: dict[str, str] = { "Origin": allowed_origin, "Access-Control-Request-Method": "GET", "Access-Control-Request-Headers": "X-Example", @@ -54,7 +55,7 @@ def test_cors_preflight_response_disallowed( self, disallowed_origin: str, client: TestClient ) -> None: """Test pre-flight response to cross-origin request from disallowed origin.""" - headers: Dict[str, str] = { + headers: dict[str, str] = { "Origin": disallowed_origin, "Access-Control-Request-Method": "GET", "Access-Control-Request-Headers": "X-Example", diff --git a/tests/test_gunicorn_conf.py b/tests/test_gunicorn_conf.py index 03f8680..09120aa 100644 --- a/tests/test_gunicorn_conf.py +++ b/tests/test_gunicorn_conf.py @@ -1,7 +1,8 @@ +from __future__ import annotations + import multiprocessing import subprocess from pathlib import Path -from typing import Optional import pytest @@ -20,7 +21,7 @@ def test_calculate_workers_default(self) -> None: assert gunicorn_conf.workers == max(cores, 2) @pytest.mark.parametrize("max_workers", (None, "1", "2", "5", "10")) - def test_calculate_workers_max(self, max_workers: Optional[str]) -> None: + def test_calculate_workers_max(self, max_workers: str | None) -> None: """Test Gunicorn worker process calculation with custom maximum.""" cores = multiprocessing.cpu_count() default = max(cores, 2) @@ -31,7 +32,7 @@ def test_calculate_workers_max(self, max_workers: Optional[str]) -> None: assert result == default @pytest.mark.parametrize("total_workers", (None, "1", "2", "5", "10")) - def test_calculate_workers_total(self, total_workers: Optional[str]) -> None: + def test_calculate_workers_total(self, total_workers: str | None) -> None: """Test Gunicorn worker process calculation with custom total.""" cores = multiprocessing.cpu_count() result = gunicorn_conf.calculate_workers(None, total_workers)