Format code and sort imports

Author: amyfromandi

services/api-v3/api/app.py

@@ -9,10 +9,10 @@
 
 import api.routes.security
 from api.database import connect_engine, dispose_engine
+from api.routes.dev import dev_router
 from api.routes.ingest import router as ingest_router
 from api.routes.object import router as object_router
 from api.routes.sources import router as sources_router
-from api.routes.dev import dev_router
 
 
 @asynccontextmanager

services/api-v3/api/models/field_site.py

@@ -1,12 +1,15 @@
-from pydantic import BaseModel
-from typing import Optional, Literal
 from datetime import datetime
 from enum import Enum
+from typing import Literal, Optional
+
+from pydantic import BaseModel
+
 
 class Location(BaseModel):
     """
     Location model representing a geographical location with latitude and longitude.
     """
+
     latitude: float
     longitude: float
     elevation: Optional[float] = None
@@ -20,49 +23,59 @@ class Location(BaseModel):
 class IdentifiedModel(BaseModel):
     id: int
 
+
 class Photo(IdentifiedModel):
     """
     A photo.
     """
+
     # URL at which the photo should be fetchable
     url: str
     width: int
     height: int
     checksum: str
 
+
 class BeddingFacing(Enum):
     upright = "upright"
     overturned = "overturned"
     unknown = "unknown"
 
+
 class PlanarOrientation(BaseModel):
     strike: float
     dip: float
     facing: BeddingFacing = BeddingFacing.upright
     notes: Optional[str] = None
     associated: list["Orientation"] = []
 
+
 class LinearOrientation(BaseModel):
     plunge: float
     trend: float
     notes: Optional[str] = None
 
+
 class Texture(BaseModel):
     name: str
 
+
 Orientation = PlanarOrientation | LinearOrientation
 
+
 class GeologicAgeInterval(IdentifiedModel):
     name: str
     t_age: Optional[float] = None
     b_age: Optional[float] = None
 
+
 class Lithology(IdentifiedModel):
     name: str
     parents: Optional[list[int]] = None
     color: Optional[str] = None
     pattern: Optional[str] = None
 
+
 class LithodemeType(Enum):
     Formation = "formation"
     Member = "member"
@@ -76,30 +89,37 @@ class LithodemeType(Enum):
     Intrusion = "intrusion"
     ...
 
+
 class LithodemeName(GeologicAgeInterval):
-    """A lithodeme or stratigraphic unit name  """
+    """A lithodeme or stratigraphic unit name"""
+
     parent: Optional[int] = None
     type: LithodemeType
     t_interval: Optional[float] = None
     b_interval: Optional[float] = None
 
+
 class RockUnit(IdentifiedModel):
     name: str
     abbreviation: Optional[str] = None
     liths: list[Lithology] = []
     age: Optional[GeologicAgeInterval] = None
     entity: Optional[LithodemeName] = None
 
+
 class Fossil(IdentifiedModel):
     description: str
     taxa: Optional[str] = None
 
+
 AnyData = Orientation | Photo | RockUnit | Texture | Lithology | Fossil
 
+
 class Observation(BaseModel):
     notes: Optional[str] = None
     data: AnyData
 
+
 class Person(IdentifiedModel):
     name: str
     email: Optional[str] = None
@@ -112,22 +132,26 @@ class Sample(IdentifiedModel):
     """
     A sample of a rock or sediment
     """
+
     name: str
     description: Optional[str] = None
     sample_type: Literal["rock", "sediment", "soil", "water"]
     igsn: Optional[str] = None
     collected: datetime
 
+
 class SocialInfo(BaseModel):
     likes: int
     comments: int
     rating: Optional[int] = None
 
+
 class FieldSite(BaseModel):
     """
     A site of with associated field observations
     """
-    id: int|str
+
+    id: int | str
     name: Optional[str] = None
     location: Location
     created: datetime
@@ -139,4 +163,3 @@ class FieldSite(BaseModel):
     social: Optional[SocialInfo] = None
     children: Optional[list["FieldSite"]] = None
     contributors: Optional[list[Person]] = None
-

services/api-v3/api/routes/dev.py

@@ -1,5 +1,5 @@
-from fastapi import APIRouter
 from api.routes.dev_routes.interchange_data import interchange_router
+from fastapi import APIRouter
 
 dev_router = APIRouter(prefix="/dev", tags=["dev"])
-dev_router.include_router(interchange_router)
+dev_router.include_router(interchange_router)

services/api-v3/api/routes/dev_routes/interchange_data.py

@@ -1,33 +1,43 @@
-from fastapi import Request, Body
-from typing import Any, Iterable, Union, List
-import httpx
+from typing import Any, Iterable, List, Union
+
 import dotenv
-from fastapi import APIRouter, Depends, HTTPException, Response, status
+import httpx
+from fastapi import APIRouter, Body, Depends, HTTPException, Request, Response, status
 from slugify import slugify
 from sqlalchemy import func, insert, select, update
 from sqlalchemy.exc import NoResultFound, NoSuchTableError
+
 dotenv.load_dotenv()
 
+from datetime import datetime, timezone
+from typing import Optional
+
 from api.database import (
     get_async_session,
     get_engine,
     get_table,
     patch_sources_sub_table,
     select_sources_sub_table,
 )
+from api.models.field_site import (
+    BeddingFacing,
+    FieldSite,
+    Location,
+    Observation,
+    Photo,
+    PlanarOrientation,
+)
 from api.query_parser import ParserException
 from api.routes.security import has_access
-from datetime import datetime, timezone
-from api.models.field_site import FieldSite, Location, Photo, PlanarOrientation, Observation, BeddingFacing
-from typing import Optional
 
 interchange_router = APIRouter(
     prefix="/interchange-data",
     tags=["interchange-data"],
     responses={404: {"description": "Not found"}},
 )
 
-#helpers
+# helpers
+
 
 def _parse_date_time(x: Optional[str]) -> Optional[datetime]:
     if not x:
@@ -40,6 +50,7 @@ def _parse_date_time(x: Optional[str]) -> Optional[datetime]:
         except Exception:
             return None
 
+
 def _to_float(v) -> Optional[float]:
     try:
         if v is None:
@@ -48,7 +59,8 @@ def _to_float(v) -> Optional[float]:
     except Exception:
         return None
 
-#normalize and require lat/lngs
+
+# normalize and require lat/lngs
 def _valid_coords(lat, lng) -> bool:
     try:
         lat = float(lat)
@@ -57,6 +69,7 @@ def _valid_coords(lat, lng) -> bool:
         return False
     return -90.0 <= lat <= 90.0 and -180.0 <= lng <= 180.0
 
+
 def _first_planar_from_spot(props) -> Optional[PlanarOrientation]:
     """Find first planar orientation with numeric strike & dip in StraboSpot props."""
     orientation = props.get("orientation_data")
@@ -69,9 +82,12 @@ def _first_planar_from_spot(props) -> Optional[PlanarOrientation]:
             strike = _to_float(item.get("strike"))
             dip = _to_float(item.get("dip"))
             if strike is not None and dip is not None:
-                return PlanarOrientation(strike=strike, dip=dip, facing=BeddingFacing.upright)
+                return PlanarOrientation(
+                    strike=strike, dip=dip, facing=BeddingFacing.upright
+                )
     return None
 
+
 def _first_planar_from_checkin(checkin) -> Optional[PlanarOrientation]:
     """Find first observation with numeric strike & dip in Rockd checkin."""
     obs = checkin.get("observations")
@@ -84,9 +100,12 @@ def _first_planar_from_checkin(checkin) -> Optional[PlanarOrientation]:
         strike = _to_float(orientation.get("strike"))
         dip = _to_float(orientation.get("dip"))
         if strike is not None and dip is not None:
-            return PlanarOrientation(strike=strike, dip=dip, facing=BeddingFacing.upright)
+            return PlanarOrientation(
+                strike=strike, dip=dip, facing=BeddingFacing.upright
+            )
     return None
 
+
 def _first_planar_from_fieldsite(fs: FieldSite) -> Optional[PlanarOrientation]:
     """Return first PlanarOrientation in FieldSite.observations."""
     for ob in fs.observations or []:
@@ -126,14 +145,16 @@ def spot_to_fieldsite(feat) -> FieldSite:
                     url=f"rockd://photo/{pid}",
                     width=int(img.get("width", 0) or 0),
                     height=int(img.get("height", 0) or 0),
-                    checksum=""
+                    checksum="",
                 )
             )
     observations: list[Observation] = []
     planar = _first_planar_from_spot(props)
     if planar:
         observations.append(Observation(data=planar))
-    created = _parse_date_time(props.get("time") or props.get("date")) or datetime.now(timezone.utc)
+    created = _parse_date_time(props.get("time") or props.get("date")) or datetime.now(
+        timezone.utc
+    )
 
     mt = props.get("modified_timestamp")
     if mt is not None:
@@ -154,8 +175,11 @@ def spot_to_fieldsite(feat) -> FieldSite:
         observations=observations,
     )
 
+
 @interchange_router.post("/spot-to-fieldsite")
-def multiple_spot_to_fieldsite(feat: Union[dict, List[dict]] = Body(...)) -> List[FieldSite]:
+def multiple_spot_to_fieldsite(
+    feat: Union[dict, List[dict]] = Body(...)
+) -> List[FieldSite]:
     """
     Accept a single FeatureCollection or a list of FeatureCollections and
     return a FieldSite for each qualifying Point feature (non-image-basemap).
@@ -199,12 +223,16 @@ def fieldsite_to_checkin(fs: FieldSite) -> dict:
         d["photo"] = fs.photos[0].id
     planar = _first_planar_from_fieldsite(fs)
     if planar:
-        d["observations"] = [{"orientation": {"strike": float(planar.strike), "dip": float(planar.dip)}}]
+        d["observations"] = [
+            {"orientation": {"strike": float(planar.strike), "dip": float(planar.dip)}}
+        ]
     return d
 
 
 @interchange_router.post("/fieldsite-to-checkin")
-def multiple_fieldsite_to_checkin(fieldsites: list[FieldSite] = Body(...)) -> list[dict]:
+def multiple_fieldsite_to_checkin(
+    fieldsites: list[FieldSite] = Body(...),
+) -> list[dict]:
     out: list[dict] = []
     for fs in fieldsites:
         try:
@@ -213,11 +241,19 @@ def multiple_fieldsite_to_checkin(fieldsites: list[FieldSite] = Body(...)) -> li
             continue
     return out
 
+
 @interchange_router.post("/spot-to-checkin")
-async def spot_to_checkin(request: Request, spot: Union[dict, List[dict]] = Body(...)) -> list[dict]:
+async def spot_to_checkin(
+    request: Request, spot: Union[dict, List[dict]] = Body(...)
+) -> list[dict]:
     """Pipeline: Spot JSON (FeatureCollection[s]) or FieldSite list -> Checkin list."""
     # If it's already a list of FieldSite-like dicts (has 'location'), skip the first hop
-    if isinstance(spot, list) and spot and isinstance(spot[0], dict) and "location" in spot[0]:
+    if (
+        isinstance(spot, list)
+        and spot
+        and isinstance(spot[0], dict)
+        and "location" in spot[0]
+    ):
         fieldsites: List[FieldSite] = spot  # already FieldSite-shaped
     else:
         # Convert FeatureCollection (or list of them) -> FieldSite list