Increase timeout, add GET to trigger fetch on demand

Author: zagura

.gitignore

@@ -2,3 +2,5 @@
 compose.override.yml
 __pycache__
 .venv/
+*.swp
+*.pyc

main.py

@@ -1,5 +1,4 @@
 import logging
-import sys
 from contextlib import asynccontextmanager
 from datetime import datetime, UTC
 from typing import Annotated, Any
@@ -39,14 +38,20 @@ def transform_card_number_to_mifare(card_number: str):
 
 class User(BaseModel):
     uid: Annotated[str, Field(validation_alias="username")]
-    mifare_card_ids: Annotated[list[str], Field(validation_alias=AliasPath("attributes", "mifareCardId"))] = []
-    unique_card_ids: Annotated[list[str], Field(validation_alias=AliasPath("attributes", "uniquecardId"))] = []
+    mifare_card_ids: Annotated[
+        list[str], Field(validation_alias=AliasPath("attributes", "mifareCardId"))
+    ] = []
+    unique_card_ids: Annotated[
+        list[str], Field(validation_alias=AliasPath("attributes", "uniquecardId"))
+    ] = []
     membership_expiration: Annotated[
         int,
-        Field(validation_alias=AliasPath("attributes", "membershipExpirationTimestamp"))
+        Field(
+            validation_alias=AliasPath("attributes", "membershipExpirationTimestamp")
+        ),
     ]
 
-    @field_validator("mifare_card_ids", "unique_card_ids", mode='plain')
+    @field_validator("mifare_card_ids", "unique_card_ids", mode="plain")
     def use_default_for_missing_cards(cls, v) -> str:
         if v is None:
             raise PydanticUseDefault()
@@ -57,7 +62,7 @@ class Settings(BaseSettings):
     authentik_token: str | None = None
     authentik_token_file: str | None = None
 
-    @model_validator(mode='after')
+    @model_validator(mode="after")
     def set_token(self) -> "Settings":
         if self.authentik_token:
             return self
@@ -96,24 +101,28 @@ def auth():
 
 async def fetch() -> list[User]:
     async with httpx.AsyncClient(
-            headers={"Authorization": f"Bearer {config.authentik_token}"},
-            timeout=30.0,
-        ) as client:
+        headers={"Authorization": f"Bearer {config.authentik_token}"},
+        timeout=60.0,
+    ) as client:
         url = (
             "https://auth.apps.hskrk.pl/api/v3/core/users/?"
-            "attributes={\"membershipExpirationTimestamp__gt\": 100}&page_size=50"
+            'attributes={"membershipExpirationTimestamp__gt": 100}&page_size=50'
         )
         response = await client.get(url)
         parsed_response = response.json()
         results = parsed_response["results"]
-        while parsed_response["pagination"]["next"]:
-            response = await client.get(f"{url}&page={parsed_response["pagination"]["next"]}")
-            parsed_response = response.json()
-            results += parsed_response["results"]
-        return [
-            User(**u)
-            for u in results
-        ]
+        next_page = parsed_response["pagination"]["next"]
+        while next_page:
+            try:
+                response = await client.get(f"{url}&page={next_page}")
+                parsed_response = response.json()
+                results += parsed_response["results"]
+                next_page = parsed_response["pagination"]["next"]
+            except httpx.ReadTimeout as t:
+                logging.warning(
+                    f"[HTTP] Timeout reached on fetching users data, page: {next_page}, {t}"
+                )
+        return [User(**u) for u in results]
 
 
 @repeat_every(seconds=300)
@@ -135,30 +144,46 @@ async def fetch_users():
         users_by_card = {
             **{
                 mifare.lower(): user
-                for user in users for mifare in user.mifare_card_ids
+                for user in users
+                for mifare in user.mifare_card_ids
             },
             **{
                 transform_card_number_to_unique(mifare).lower(): user
-                for user in users for mifare in user.mifare_card_ids
+                for user in users
+                for mifare in user.mifare_card_ids
             },
             **{
                 unique.lower(): user
-                for user in users for unique in user.unique_card_ids
+                for user in users
+                for unique in user.unique_card_ids
             },
             **{
                 transform_card_number_to_mifare(unique).lower(): user
-                for user in users for unique in user.unique_card_ids
+                for user in users
+                for unique in user.unique_card_ids
             },
         }
-        logging.debug(f"Fetched {len(users)} users ({len(users_by_card)} cards) from Authentik")
+        logging.debug(
+            f"Fetched {len(users)} users ({len(users_by_card)} cards) from Authentik"
+        )
 
 
 @app.get("/users/-/stats")
 async def get_user_stats():
     return {
-        "last_success_run": users_last_success_run.isoformat().replace("+00:00", "Z") if users_last_success_run else None,
-        "last_failed_run": users_last_failed_run.isoformat().replace("+00:00", "Z") if users_last_failed_run else None,
-        "last_failed_reason": str(users_last_failed_reason) if users_last_failed_reason else None,
+        "last_success_run": (
+            users_last_success_run.isoformat().replace("+00:00", "Z")
+            if users_last_success_run
+            else None
+        ),
+        "last_failed_run": (
+            users_last_failed_run.isoformat().replace("+00:00", "Z")
+            if users_last_failed_run
+            else None
+        ),
+        "last_failed_reason": (
+            str(users_last_failed_reason) if users_last_failed_reason else None
+        ),
         "users": {
             "count": len(users),
         },
@@ -178,6 +203,37 @@ async def get_user_by_card(card_id: str):
     return user
 
 
+@app.get("/user/-/sync")
+async def sync_on_demand():
+    global users_last_success_run
+    current_run = datetime.now(tz=UTC)
+    diff_seconds = (current_run - users_last_success_run).seconds
+    if diff_seconds < 30:
+        logging.debug("Skipping fetching users, data fresh")
+        return {
+            "cached": True,
+            "last_success_run": (
+                users_last_success_run.isoformat().replace("+00:00", "Z")
+                if users_last_success_run
+                else None
+            ),
+            "users": {
+                "count": len(users),
+            },
+            "last_failed_run": (
+                users_last_failed_run.isoformat().replace("+00:00", "Z")
+                if users_last_failed_run
+                else None
+            ),
+        }
+    else:
+        logging.debug(
+            "Last successful fetch {diff_seconds}s before. Fetching on demand"
+        )
+        fetch_users()
+        return get_user_stats()
+
+
 @app.websocket("/ws")
 async def websocket_endpoint(websocket: WebSocket):
     logging.debug("[WS] New connection")
@@ -199,7 +255,9 @@ async def websocket_endpoint(websocket: WebSocket):
                             {"status": "error", "error": "user not found"}
                         )
                     else:
-                        logging.info("[WS] User %s (%s) found", user.uid, kwargs.get("card_id"))
+                        logging.info(
+                            "[WS] User %s (%s) found", user.uid, kwargs.get("card_id")
+                        )
                         await websocket.send_json(
                             {"status": "ok", "object": user.model_dump()}
                         )