Merge pull request #38 from a-luna:refactor-api-settings_20240126

Minor bug fixes and refactorings

Dockerfile

@@ -21,7 +21,6 @@ ENV REDIS_PW=${REDIS_PW}
 ENV RATE_LIMIT_PER_PERIOD=${RATE_LIMIT_PER_PERIOD}
 ENV RATE_LIMIT_PERIOD_SECONDS=${RATE_LIMIT_PERIOD_SECONDS}
 ENV RATE_LIMIT_BURST=${RATE_LIMIT_BURST}
-ENV TEST_HEADER=${TEST_HEADER}
 
 WORKDIR /code
 
@@ -36,7 +35,6 @@ RUN echo "REDIS_PW=$REDIS_PW" >> /code/.env
 RUN echo "RATE_LIMIT_PER_PERIOD=$RATE_LIMIT_PER_PERIOD" >> /code/.env
 RUN echo "RATE_LIMIT_PERIOD_SECONDS=$RATE_LIMIT_PERIOD_SECONDS" >> /code/.env
 RUN echo "RATE_LIMIT_BURST=$RATE_LIMIT_BURST" >> /code/.env
-RUN echo "TEST_HEADER=$TEST_HEADER" >> /code/.env
 
 RUN pip install -U pip setuptools wheel
 COPY ./requirements.txt /code/requirements.txt

app/api/api_v1/endpoints/characters.py

@@ -12,7 +12,6 @@
 from app.api.api_v1.dependencies.filter_param_matcher import filter_param_matcher
 from app.api.api_v1.endpoints.util import get_character_details
 from app.api.api_v1.pagination import paginate_search_results
-from app.config.api_settings import get_settings
 from app.data.cache import cached_data
 from app.data.encoding import get_codepoint_string
 from app.db.session import DBSession, get_session
@@ -38,7 +37,7 @@ def list_all_unicode_characters(
 ):
     (start, stop) = get_char_list_endpoints(list_params, block)
     return {
-        "url": f"{get_settings().API_VERSION}/characters",
+        "url": f"{db_ctx.api_settings.API_VERSION}/characters",
         "has_more": stop <= block.finish,
         "data": [get_character_details(db_ctx, codepoint, []) for codepoint in range(start, stop)],
     }
@@ -53,7 +52,7 @@ def search_unicode_characters_by_name(
     db_ctx: Annotated[DBSession, Depends(get_session)],
     search_parameters: Annotated[CharacterSearchParameters, Depends()],
 ):
-    response_data = {"url": f"{get_settings().API_VERSION}/characters/search", "query": search_parameters.name}
+    response_data = {"url": f"{db_ctx.api_settings.API_VERSION}/characters/search", "query": search_parameters.name}
     search_results = cached_data.search_characters_by_name(search_parameters.name, search_parameters.min_score)
     return get_paginated_character_list(
         db_ctx,
@@ -80,7 +79,7 @@ def filter_unicode_characters(
             detail="No filter settings were specified in the request.",
         )
     response_data = {
-        "url": f"{get_settings().API_VERSION}/characters/filter",
+        "url": f"{db_ctx.api_settings.API_VERSION}/characters/filter",
         "filter_settings": filter_parameters.settings,
     }
     codepoints = db_ctx.filter_all_characters(filter_parameters)

app/config/api_settings.py

@@ -4,11 +4,29 @@
 from dataclasses import dataclass, field
 from datetime import timedelta
 from pathlib import Path
+from typing import TypedDict
 
 import app.db.models as db
 from app.config.dotenv_file import read_dotenv_file
 from app.data.constants import UNICODE_PLANES_DEFAULT, UNICODE_VERSION_RELEASE_DATES
 
+
+class ApiSettingsDict(TypedDict):
+    ENV: str
+    UNICODE_VERSION: str
+    PROJECT_NAME: str
+    API_VERSION: str
+    REDIS_PW: str
+    REDIS_HOST: str
+    REDIS_PORT: int
+    REDIS_DB: int
+    RATE_LIMIT_PER_PERIOD: int
+    RATE_LIMIT_PERIOD_SECONDS: timedelta
+    RATE_LIMIT_BURST: int
+    SERVER_NAME: str
+    SERVER_HOST: str
+
+
 UNICODE_ORG_ROOT = "https://www.unicode.org/Public"
 UNICODE_XML_FOLDER = "ucdxml"
 HTTP_BUCKET_URL = "https://unicode-api.us-southeast-1.linodeobjects.com"
@@ -48,7 +66,6 @@ class UnicodeApiSettings:
     RATE_LIMIT_BURST: int
     SERVER_NAME: str
     SERVER_HOST: str
-    CACHE_HEADER: str
     API_ROOT: str = field(init=False, default="")
     ROOT_FOLDER: Path = field(init=False)
     APP_FOLDER: Path = field(init=False)
@@ -85,7 +102,7 @@ def __post_init__(self) -> None:
         json_folder = version_folder.joinpath("json")
         csv_folder = version_folder.joinpath("csv")
 
-        self.API_ROOT = DEV_API_ROOT if "PROD" not in self.ENV else PROD_API_ROOT
+        self.API_ROOT = DEV_API_ROOT if self.is_dev else PROD_API_ROOT
         self.ROOT_FOLDER = ROOT_FOLDER
         self.APP_FOLDER = ROOT_FOLDER.joinpath("app")
         self.DATA_FOLDER = data_folder
@@ -182,7 +199,7 @@ def init_data_folders(self) -> None:  # pragma: no cover
 
 def get_api_settings() -> UnicodeApiSettings:  # pragma: no cover
     env_vars = read_dotenv_file(DOTENV_FILE)
-    settings = {
+    settings: ApiSettingsDict = {
         "ENV": env_vars.get("ENV", "DEV"),
         "UNICODE_VERSION": env_vars.get("UNICODE_VERSION", get_latest_unicode_version()),
         "PROJECT_NAME": "Unicode API",
@@ -196,13 +213,12 @@ def get_api_settings() -> UnicodeApiSettings:  # pragma: no cover
         "RATE_LIMIT_BURST": int(env_vars.get("RATE_LIMIT_BURST", "10")),
         "SERVER_NAME": "unicode-api.aaronluna.dev",
         "SERVER_HOST": PROD_API_ROOT,
-        "CACHE_HEADER": "X-UnicodeAPI-Cache",
     }
     return UnicodeApiSettings(**settings)
 
 
 def get_test_settings() -> UnicodeApiSettings:
-    settings = {
+    settings: ApiSettingsDict = {
         "ENV": "TEST",
         "UNICODE_VERSION": "15.0.0",
         "PROJECT_NAME": "Test Unicode API",
@@ -216,14 +232,12 @@ def get_test_settings() -> UnicodeApiSettings:
         "RATE_LIMIT_BURST": 1,
         "SERVER_NAME": "",
         "SERVER_HOST": "",
-        "CACHE_HEADER": "",
     }
     return UnicodeApiSettings(**settings)
 
 
 def get_settings() -> UnicodeApiSettings:
-    env = os.environ.get("ENV", "DEV")
-    settings = get_test_settings() if "TEST" in env else get_api_settings()
+    settings = get_test_settings() if "TEST" in os.environ.get("ENV", "DEV") else get_api_settings()
     logger = logging.getLogger("app.api")
     logger.debug(settings.api_settings_report)
     logger.debug(settings.rate_limit_settings_report)

app/core/logging.py

@@ -13,7 +13,7 @@
     "formatters": {
         "default": {
             "()": "app.core.logging.DefaultFormatter",
-            "fmt": "%(levelprefix)s %(message)s",
+            "fmt": "%(levelprefix)s [unicode-api] %(message)s",
             "use_colors": True,
         },
     },

app/core/rate_limit.py

@@ -81,7 +81,7 @@ def is_exceeded(self, request: Request) -> Result[None]:
         Adapted for Python from this article:
         https://vikas-kumar.medium.com/rate-limiting-techniques-245c3a5e9cad
         """
-        if self.rate_limit_not_required(request):
+        if not self.apply_rate_limit_to_request(request):
             return Result.Ok()
         client_ip = request.client.host if request.client else "localhost"
         arrived_at = self.redis.time()
@@ -100,10 +100,10 @@ def is_exceeded(self, request: Request) -> Result[None]:
         except LockError:  # pragma: no cover
             return self.lock_error(client_ip)
 
-    def rate_limit_not_required(self, request: Request):
-        if self.settings.is_prod or self.settings.is_dev:  # pragma: no cover
-            return request_origin_is_internal(request) and requested_route_is_not_rate_limited(request)
-        return not rate_limit_feature_is_under_test(request)
+    def apply_rate_limit_to_request(self, request: Request):
+        if self.settings.is_test:
+            return enable_rate_limit_feature_for_test(request)
+        return request_origin_is_external(request) and requested_route_is_rate_limited(request)
 
     def get_allowed_at(self, tat: float) -> float:
         return (dtaware_fromtimestamp(tat) - self.delay_tolerance_ms).timestamp()
@@ -135,7 +135,7 @@ def lock_error(self, client) -> Result[None]:  # pragma: no cover
         return Result.Fail(error)
 
 
-def rate_limit_feature_is_under_test(request: Request) -> bool:
+def enable_rate_limit_feature_for_test(request: Request) -> bool:
     if "x-verify-rate-limiting" in request.headers:
         return request.headers["x-verify-rate-limiting"] == "true"
     if "access-control-request-headers" in request.headers:  # pragma: no cover
@@ -145,16 +145,16 @@ def rate_limit_feature_is_under_test(request: Request) -> bool:
     return False  # pragma: no cover
 
 
-def request_origin_is_internal(request: Request) -> bool:  # pragma: no cover
-    if "localhost" in request.client.host:
-        return True
+def request_origin_is_external(request: Request) -> bool:
+    if request.client.host in ["localhost", "127.0.0.1", "testserver"]:
+        return False
     if "sec-fetch-site" in request.headers:
-        return request.headers["sec-fetch-site"] == "same-site"
-    return False
+        return request.headers["sec-fetch-site"] != "same-site"
+    return True
 
 
-def requested_route_is_not_rate_limited(request: Request):  # pragma: no cover
-    return not RATE_LIMIT_ROUTE_REGEX.search(request.url.path)
+def requested_route_is_rate_limited(request: Request):
+    return RATE_LIMIT_ROUTE_REGEX.search(request.url.path)
 
 
 def get_time_portion(ts: float) -> str:

app/core/util.py

@@ -7,9 +7,11 @@
 
 
 def get_unicode_version_release_date(version: str) -> str:
-    if release_date := UNICODE_VERSION_RELEASE_DATES.get(version, None):
-        return release_date.strftime(DATE_MONTH_NAME)
-    return ""
+    return (
+        release_date.strftime(DATE_MONTH_NAME)
+        if (release_date := UNICODE_VERSION_RELEASE_DATES.get(version, None))
+        else ""
+    )
 
 
 def make_tzaware(dt: datetime, use_tz: tzinfo | None = None, localize: bool = True) -> datetime:
@@ -57,9 +59,9 @@ def format_timedelta_str(td: timedelta, precise: bool = True) -> str:
     return duration
 
 
-def get_time_until_timestamp(ts: float, precise: bool = True) -> str:
-    return get_duration_between_timestamps(datetime.now().timestamp(), ts, precise)
+def get_time_until_timestamp(ts: float) -> timedelta:
+    return get_duration_between_timestamps(datetime.now().timestamp(), ts)
 
 
-def get_duration_between_timestamps(ts1: float, ts2: float, precise: bool = True) -> str:
+def get_duration_between_timestamps(ts1: float, ts2: float) -> timedelta:
     return dtaware_fromtimestamp(ts2) - dtaware_fromtimestamp(ts1)

app/data/scripts/script_types.py

@@ -3,6 +3,4 @@
 CharDetailsDict = dict[str, bool | int | str]
 BlockOrPlaneDetailsDict = dict[str, int | str]
 AllParsedUnicodeData = tuple[list[BlockOrPlaneDetailsDict], list[BlockOrPlaneDetailsDict], list[CharDetailsDict]]
-UnicodeModel = (
-    type[db.UnicodePlane] | type[db.UnicodeBlock] | type[db.UnicodeCharacter] | type[db.UnicodeCharacterUnihan]
-)
+UnicodeModel = db.UnicodePlane | db.UnicodeBlock | db.UnicodeCharacter | db.UnicodeCharacterUnihan

app/data/scripts/update_all_data.py

@@ -51,14 +51,14 @@ def update_all_data() -> Result[None]:
 
 def get_xml_unicode_database(config: UnicodeApiSettings) -> Result[Path]:
     spinner = Spinner()
-    get_xml_result = download_xml_unicode_database(config)
-    if get_xml_result.failure or not get_xml_result.value:
+    result = download_xml_unicode_database(config)
+    if result.failure or not result.value:
         spinner.start("")
         spinner.failed("Download failed! Please check the internet connection.")
-        return get_xml_result
+        return result
     spinner.start("")
     spinner.successful(f"Successfully downloaded Unicode XML Database v{config.UNICODE_VERSION}!")
-    xml_file = get_xml_result.value
+    xml_file = result.value
     return Result.Ok(xml_file)
 
 

app/db/session.py

@@ -3,9 +3,10 @@
 from sqlalchemy.engine import Engine
 from sqlmodel import Session
 
-import app.db.procs.filter_characters as proc_filter_all_characters
-import app.db.procs.get_char_details as proc_get_character_properties
-import app.db.procs.get_unicode_versions as proc_get_all_unicode_versions
+import app.db.procs.filter_characters as proc_filter
+import app.db.procs.get_char_details as proc_char
+import app.db.procs.get_unicode_versions as proc_ver
+from app.config.api_settings import get_settings
 from app.db.engine import engine
 from app.schemas.enums import CharPropertyGroup
 
@@ -22,14 +23,15 @@ class DBSession:
     def __init__(self, session: Session, engine: Engine):
         self.session = session
         self.engine = engine
+        self.api_settings = get_settings()
 
     def all_unicode_versions(self):
-        return proc_get_all_unicode_versions.get_all_unicode_versions(self.session)
+        return proc_ver.get_all_unicode_versions(self.session)
 
     def get_character_properties(
         self, codepoint: int, show_props: list[CharPropertyGroup] | None, verbose: bool
     ) -> dict[str, Any]:
-        return proc_get_character_properties.get_character_properties(self.engine, codepoint, show_props, verbose)
+        return proc_char.get_character_properties(self.engine, codepoint, show_props, verbose)
 
     def filter_all_characters(self, filter_params: "FilterParameters") -> list[int]:
-        return proc_filter_all_characters.filter_all_characters(self.session, filter_params)
+        return proc_filter.filter_all_characters(self.session, filter_params)

app/schemas/enums/char_filter_flags.py

@@ -55,7 +55,7 @@ def __repr__(self) -> str:
 
     @property
     def normalized(self) -> str:
-        return normalize_string_lm3(self.name)
+        return normalize_string_lm3(str(self.name))
 
     @property
     def flag_name(self) -> str:

app/tests/test_rate_limiting/test_rate_limiting.py

@@ -1,23 +1,6 @@
-import os
-
-import pytest
-from fastapi.testclient import TestClient
-
 from app.tests.test_rate_limiting.data import PLANE_0
 
 
-@pytest.fixture
-def client():
-    os.environ["ENV"] = "TEST"
-    from app.main import app
-
-    with TestClient(app) as client:
-        headers = {}
-        headers["x-verify-rate-limiting"] = "true"
-        client.headers = headers
-        yield client
-
-
 def test_rate_limiting(client):
     response = client.get("/v1/planes/0")
     assert response.status_code == 200

docker-bake.hcl

@@ -23,6 +23,5 @@ target "unicode-api" {
         RATE_LIMIT_PER_PERIOD="50"
         RATE_LIMIT_PERIOD_SECONDS="60"
         RATE_LIMIT_BURST="10"
-        TEST_HEADER="X-UnicodeAPI-Test"
     }
 }

requirements-lock.txt

@@ -7,7 +7,7 @@ asttokens==2.4.1
 async-timeout==4.0.3
 attrs==23.2.0
 backcall==0.2.0
-black==23.12.1
+black==24.1.0
 cachetools==5.3.2
 certifi==2023.11.17
 chardet==5.2.0