Added API for fetching external database, removed parameters structure,

spoolman/externaldb.py

@@ -3,12 +3,15 @@
 import datetime
 import logging
 import os
+from pathlib import Path
 from typing import Optional
 
 import hishel
-from pydantic import BaseModel, ConfigDict, Extra, Field
+from pydantic import BaseModel, Field
 from scheduler.asyncio.scheduler import Scheduler
 
+from spoolman import filecache
+
 DEFAULT_EXTERNAL_DB_URL = "https://donkie.github.io/SpoolmanDB/"
 DEFAULT_SYNC_INTERVAL = 3600
 
@@ -18,26 +21,18 @@ cache_storage = hishel.AsyncFileStorage()
 logger = logging.getLogger(__name__)
 
 
-class ExternalFilamentParameters(BaseModel):
-    model_config = ConfigDict(
-        extra=Extra.allow,
-    )
-
-    extruder_temp: Optional[float] = Field(default=None)
-    bed_temp: Optional[float] = Field(default=None)
-
-
 class ExternalFilament(BaseModel):
-    id: str = Field()
-    manufacturer: str = Field()
-    name: str = Field()
-    material: str = Field()
-    density: Optional[float] = Field(default=None)
-    weight: float = Field(default=None)
-    spool_weight: Optional[float] = Field(default=None)
-    diameter: float = Field(default=None)
-    color_hex: str = Field()
-    parameters: ExternalFilamentParameters = Field(ExternalFilamentParameters())
+    id: str = Field(description="A unique ID for this filament.", example="polymaker_pla_polysonicblack_1000_175")
+    manufacturer: str = Field(description="Filament manufacturer.", example="Polymaker")
+    name: str = Field(description="Filament name.", example="Polysonic\u2122 Black")
+    material: str = Field(description="Filament material.", example="PLA")
+    density: Optional[float] = Field(default=None, description="Density in g/cm3.", example=1.23)
+    weight: float = Field(description="Net weight of a single spool.", example=1000)
+    spool_weight: Optional[float] = Field(default=None, description="Weight of an empty spool.", example=140)
+    diameter: float = Field(description="Filament in mm.", example=1.75)
+    color_hex: str = Field(description="Filament color code in hex format.", example="#2c3232")
+    extruder_temp: Optional[int] = Field(default=None, description="Extruder/nozzle temperature in °C.", example=210)
+    bed_temp: Optional[int] = Field(default=None, description="Bed temperature in °C.", example=50)
 
 
 class ExternalFilamentsFile(BaseModel):
@@ -45,27 +40,14 @@ class ExternalFilamentsFile(BaseModel):
 
 
 class ExternalMaterial(BaseModel):
-    material: str = Field()
-    density: float = Field()
+    material: str = Field(example="PLA")
+    density: float = Field(example=1.24)
 
 
 class ExternalMaterialsFile(BaseModel):
     __root__: list[ExternalMaterial]
 
 
-class ExternalDB:
-    filaments: list[ExternalFilament]
-    materials: list[ExternalMaterial]
-
-    def __init__(self) -> None:
-        """Initialize the ExternalDB."""
-        self.filaments = []
-        self.materials = []
-
-
-_externaldb = ExternalDB()
-
-
 def get_external_db_url() -> str:
     """Get the external database URL from environment variables. Defaults to DEFAULT_EXTERNAL_DB_URL."""
     return os.getenv("EXTERNAL_DB_URL", DEFAULT_EXTERNAL_DB_URL).strip("/") + "/"
@@ -87,31 +69,46 @@ async def _download_file(url: str) -> bytes:
         return response.read()
 
 
-def _parse_filaments_from_bytes(data: bytes) -> list[ExternalFilament]:
+def _parse_filaments_from_bytes(data: bytes) -> ExternalFilamentsFile:
     """Parse a bytes string into a list of ExternalFilament objects."""
-    return ExternalFilamentsFile.parse_raw(data).__root__
+    return ExternalFilamentsFile.parse_raw(data)
 
 
-def _parse_materials_from_bytes(data: bytes) -> list[ExternalMaterial]:
+def _parse_materials_from_bytes(data: bytes) -> ExternalMaterialsFile:
     """Parse a bytes string into a list of ExternalMaterial objects."""
-    return ExternalMaterialsFile.parse_raw(data).__root__
+    return ExternalMaterialsFile.parse_raw(data)
 
 
-def get_external_db() -> ExternalDB:
-    """Get the external database."""
-    return _externaldb
+def _write_to_local_cache(filename: str, data: bytes) -> None:
+    """Write data to the local cache."""
+    filecache.update_file(filename, data)
+
+
+def get_filaments_file() -> Path:
+    """Get the path to the filaments file."""
+    return filecache.get_file("filaments.json")
+
+
+def get_materials_file() -> Path:
+    """Get the path to the materials file."""
+    return filecache.get_file("materials.json")
 
 
 async def _sync() -> None:
     logger.info("Syncing external DB.")
 
     url = get_external_db_url()
-    _externaldb.filaments = _parse_filaments_from_bytes(await _download_file(url + "filaments.json"))
-    _externaldb.materials = _parse_materials_from_bytes(await _download_file(url + "materials.json"))
+
+    filaments = _parse_filaments_from_bytes(await _download_file(url + "filaments.json"))
+    materials = _parse_materials_from_bytes(await _download_file(url + "materials.json"))
+
+    _write_to_local_cache("filaments.json", filaments.json().encode())
+    _write_to_local_cache("materials.json", materials.json().encode())
+
     logger.info(
         "External DB synced. Filaments: %d, Materials: %d",
-        len(_externaldb.filaments),
-        len(_externaldb.materials),
+        len(filaments.__root__),
+        len(materials.__root__),
     )
 
 
@@ -121,6 +118,10 @@ def schedule_tasks(scheduler: Scheduler) -> None:
     Args:
         scheduler: The scheduler to use for scheduling tasks.
     """
+    if len(get_external_db_url().strip()) == 0:
+        logger.info("External DB URL is empty. Skipping sync.")
+        return
+
     logger.info("Scheduling external DB sync.")
 
     # Run once on startup