Skip to content

opensampl.server.backend.main

API Configuration to Indirectly interact with the database

ProbeMetadataPayload

Bases: BaseModel

Probe Metadata Payload Model

Source code in opensampl/server/backend/main.py 
47
48
49
50
51
52
class ProbeMetadataPayload(BaseModel):
    """Probe Metadata Payload Model"""

    vendor: VendorType
    probe_key: ProbeKey
    data: dict[str, Any]

TimeDataPoint

Bases: BaseModel

Time Data Model

Source code in opensampl/server/backend/main.py 
32
33
34
35
36
class TimeDataPoint(BaseModel):
    """Time Data Model"""

    time: str
    value: float

WriteTablePayload

Bases: BaseModel

Write Table Payload Model

Source code in opensampl/server/backend/main.py 
39
40
41
42
43
44
class WriteTablePayload(BaseModel):
    """Write Table Payload Model"""

    table: str
    data: dict[str, Any]
    if_exists: load_data.conflict_actions = "update"

check_log_level(api_key=Depends(require_api_key()))

Check which log levels are visible in backend container

Source code in opensampl/server/backend/main.py 
169
170
171
172
173
174
175
176
177
@app.get("/checkloglevel")
def check_log_level(api_key: str = Depends(require_api_key())):
    """Check which log levels are visible in backend container"""
    logger.debug("Debug test")
    logger.info("Info test")
    logger.warning("Warning test")
    logger.error("Error test")
    current_level = next(iter(logger._core.handlers.values()))["level"].name  # noqa: SLF001
    return {"loglevel": current_level}

create_new_tables(create_schema=True, api_key=Depends(require_api_key()), session=Depends(get_db))

Update DB based on ORM Tables

Source code in opensampl/server/backend/main.py 
304
305
306
307
308
309
310
311
312
313
314
315
316
317
318
319
320
@app.get("/create_new_tables")
def create_new_tables(
    create_schema: bool = True, api_key: str = Depends(require_api_key()), session: Session = Depends(get_db)
):
    """Update DB based on ORM Tables"""
    try:
        load_data.create_new_tables(create_schema=create_schema, session=session)
        return JSONResponse(content={"message": "Succeeded in creating any new tables"}, status_code=200)
    except SQLAlchemyError as e:
        logger.error(f"SQLAlchemy error: {e}")
        return JSONResponse(content={"message": f"Database error: {e}"}, status_code=500)
    except json.JSONDecodeError as e:
        logger.error(f"JSON decode error: {e}")
        return JSONResponse(content={"message": f"Invalid JSON data: {e}"}, status_code=400)
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return JSONResponse(content={"message": f"Failed to load JSON into database: {e}"}, status_code=500)

docs_redirect() async

Redirect bare url to docs

Source code in opensampl/server/backend/main.py 
155
156
157
158
@app.get("/", include_in_schema=False)
async def docs_redirect():
    """Redirect bare url to docs"""
    return RedirectResponse(url="/docs")

generate_api_key(expire_after=None, api_key=Depends(require_api_key(bootstrap=True)), session=Depends(get_db))

Generate new API key in the database

Source code in opensampl/server/backend/main.py 
323
324
325
326
327
328
329
330
331
332
333
334
335
336
337
338
339
340
341
342
343
344
345
@app.get("/gen_api_key")
def generate_api_key(
    expire_after: int | None = None,
    api_key: str | None = Depends(require_api_key(bootstrap=True)),
    session: Session = Depends(get_db),
):
    """Generate new API key in the database"""
    try:
        new_key = APIAccessKey()
        new_key.generate_key()
        if expire_after:
            new_key.expires_at = datetime.now(tz=UTC) + timedelta(days=expire_after)

        session.add(new_key)

        session.commit()
        return JSONResponse(content={"message": "Succeeded in creating new access key"}, status_code=200)
    except SQLAlchemyError as e:
        logger.error(f"SQLAlchemy error: {e}")
        return JSONResponse(content={"message": f"Database error: {e}"}, status_code=500)
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return JSONResponse(content={"message": f"Failed to create new access key: {e}"}, status_code=500)

get_db()

Get database session

Source code in opensampl/server/backend/main.py 
128
129
130
131
132
133
134
135
def get_db():
    """Get database session"""
    Session = sessionmaker(bind=engine)  # noqa: N806
    try:
        session = Session()
        yield session
    finally:
        session.close()

get_keys()

Get active API keys

Source code in opensampl/server/backend/main.py 
 88
 89
 90
 91
 92
 93
 94
 95
 96
 97
 98
 99
100
101
102
103
104
105
106
def get_keys():
    """Get active API keys"""
    env_keys = os.getenv("API_KEYS", "").strip()
    keys = [k.strip() for k in env_keys.split(",") if k.strip()]
    if keys:
        logger.debug("api access keys loaded from env")
        return keys
    try:
        Session = sessionmaker(bind=engine)  # noqa: N806
        with Session() as session:
            now = datetime.now(tz=UTC)
            stmt = select(APIAccessKey.key).where(or_(APIAccessKey.expires_at.is_(None), APIAccessKey.expires_at > now))
            result = session.execute(stmt)
            keys = [row[0] for row in result.all()]
            logger.debug("api access keys loaded from db")
            return keys
    except Exception as e:
        logger.debug(f"exception attempting to load api access keys from db: {e}")
        return []

healthcheck()

Ensure the api is accepting queries

Source code in opensampl/server/backend/main.py 
348
349
350
351
@app.get("/healthcheck")
def healthcheck():
    """Ensure the api is accepting queries"""
    return {"status": "OK"}

healthcheck_db()

Ensure the db is accepting connections

Source code in opensampl/server/backend/main.py 
354
355
356
357
358
359
360
361
362
363
@app.get("/healthcheck_database")
def healthcheck_db():
    """Ensure the db is accepting connections"""
    try:
        with engine.connect() as connection:
            connection.execute(text("SELECT 1"))
    except SQLAlchemyError as e:
        return JSONResponse(content={"message": f"Database connection error: {e!s}"}, status_code=503)
    else:
        return {"status": "OK"}

healthcheck_metadata()

Ensure that the database exists AND the expected format is present

Source code in opensampl/server/backend/main.py 
366
367
368
369
370
371
372
373
374
375
376
377
378
379
380
381
382
383
384
@app.get("/healthcheck_metadata")
def healthcheck_metadata():
    """Ensure that the database exists AND the expected format is present"""
    # eventually, we want to make the schema configurable through environment variables
    # for now, we have it hard coded too many places. So this is a small step towards that goal
    SCHEMA = "castdb"  # noqa: N806

    try:
        with engine.connect() as connection:
            result = connection.execute(
                text("SELECT schema_name FROM information_schema.schemata WHERE schema_name = :schema;"),
                {"schema": SCHEMA},
            )
            schema_exists = result.fetchone() is not None
        if schema_exists:
            return {"status": "OK"}
        return JSONResponse(status_code=500, content={"message": f"Expected schema '{SCHEMA}' does not exist"})
    except SQLAlchemyError as e:
        return JSONResponse(content={"message": f"Database connection error: {e!s}"}, status_code=503)

load_probe_metadata(payload, api_key=Depends(require_api_key()), session=Depends(get_db))

Load metadata for given probe

Source code in opensampl/server/backend/main.py 
273
274
275
276
277
278
279
280
281
282
283
284
285
286
287
288
289
290
291
292
293
294
295
296
297
298
299
300
301
@app.post("/load_probe_metadata")
def load_probe_metadata(
    payload: ProbeMetadataPayload, api_key: str = Depends(require_api_key()), session: Session = Depends(get_db)
):
    """Load metadata for given probe"""
    logger.debug(f"Received payload: {payload.model_dump()}")

    try:
        load_data.load_probe_metadata(
            vendor=payload.vendor, probe_key=payload.probe_key, data=payload.data, session=session
        )
        logger.debug(
            f"Successfully wrote to {ProbeMetadata.__tablename__} and {payload.vendor.metadata_table}: {payload.data}"
        )
        return JSONResponse(content={"message": f"Succeeded loaded metadata for {payload.probe_key}"}, status_code=200)
    except IntegrityError as e:
        session.rollback()
        if isinstance(e.orig, psycopg2.errors.UniqueViolation):
            return JSONResponse(content={"message": f"Unique violation error: {e}"}, status_code=409)
        return JSONResponse(content={"message": f"Integrity error: {e}"}, status_code=500)
    except SQLAlchemyError as e:
        logger.error(f"SQLAlchemy error: {e}")
        return JSONResponse(content={"message": f"Database error: {e}"}, status_code=500)
    except json.JSONDecodeError as e:
        logger.error(f"JSON decode error: {e}")
        return JSONResponse(content={"message": f"Invalid JSON data: {e}"}, status_code=400)
    except Exception as e:
        logger.exception(f"Unexpected error: {e}")
        return JSONResponse(content={"message": f"Failed to load JSON into database: {e}"}, status_code=500)

load_time_data(probe_key_str=Form(...), metric_type_str=Form(None), reference_type_str=Form(None), compound_key_str=Form(None), file=File(...), api_key=Depends(require_api_key()), session=Depends(get_db)) async

Load provided data for given probe

Source code in opensampl/server/backend/main.py 
204
205
206
207
208
209
210
211
212
213
214
215
216
217
218
219
220
221
222
223
224
225
226
227
228
229
230
231
232
233
234
235
236
237
238
239
240
241
242
243
244
245
246
247
248
249
250
251
252
253
254
255
256
257
258
259
260
261
262
263
264
265
266
267
268
269
270
@app.post("/load_time_data")
async def load_time_data(  # noqa: PLR0912, C901
    probe_key_str: str = Form(...),
    metric_type_str: str | None = Form(None),
    reference_type_str: str | None = Form(None),
    compound_key_str: str | None = Form(None),
    file: UploadFile = File(...),
    api_key: str = Depends(require_api_key()),
    session: Session = Depends(get_db),
):
    """Load provided data for given probe"""
    try:
        probe_key = ProbeKey(**json.loads(probe_key_str))

        if metric_type_str is not None:
            metric_type_dict = json.loads(metric_type_str)
            metric_type = MetricType(**metric_type_dict)
        else:
            metric_type = METRICS.UNKNOWN

        if reference_type_str is not None:
            reference_type_dict = json.loads(reference_type_str)
            if "reference_table" in reference_type_dict:
                reference_type = CompoundReferenceType(**reference_type_dict)
            else:
                reference_type = ReferenceType(**reference_type_dict)
        else:
            reference_type = REF_TYPES.UNKNOWN

        compound_key = None if compound_key_str is None else json.loads(compound_key_str)

        content = await file.read()
        df = pd.read_csv(io.BytesIO(content))
        logger.info(df.head())
        # Convert time strings back to datetime
        df["time"] = pd.to_datetime(df["time"])

        # Use the same load_time_data function as before
        load_data.load_time_data(
            probe_key=probe_key,
            metric_type=metric_type,
            reference_type=reference_type,
            compound_key=compound_key,
            data=df,
            session=session,
        )

        return JSONResponse(content={"message": f"Successfully loaded {len(df)} data points"}, status_code=200)
    except IntegrityError as e:
        if session:
            session.rollback()
            session.close()
        if isinstance(e.orig, psycopg2.errors.UniqueViolation):
            return JSONResponse(content={"message": f"Unique violation error: {e}"}, status_code=409)
        return JSONResponse(content={"message": f"Integrity error: {e}"}, status_code=500)
    except SQLAlchemyError as e:
        logger.error(f"Database error: {e}")
        if session:
            session.rollback()
            session.close()
        raise HTTPException(status_code=500, detail=f"Database error: {e!s}") from e
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        if session:
            session.rollback()
            session.close()
        raise HTTPException(status_code=500, detail=f"Error processing time series data: {e!s}") from e

metrics()

Expose Prometheus metrics.

Source code in opensampl/server/backend/main.py 
387
388
389
390
@app.get("/metrics", include_in_schema=False)
def metrics():
    """Expose Prometheus metrics."""
    return Response(content=generate_latest(), media_type=CONTENT_TYPE_LATEST)

metrics_middleware(request, call_next) async

Middleware to track request metrics.

Source code in opensampl/server/backend/main.py 
138
139
140
141
142
143
144
145
146
147
148
149
150
151
@app.middleware("http")
async def metrics_middleware(request: Request, call_next: Callable) -> Response:
    """Middleware to track request metrics."""
    if request.url.path in EXCLUDED_PATHS:
        return await call_next(request)
    start_time = time.time()
    response: Response = await call_next(request)
    duration = time.time() - start_time

    REQUEST_COUNT.labels(method=request.method, endpoint=request.url.path, http_status=response.status_code).inc()

    REQUEST_LATENCY.labels(method=request.method, endpoint=request.url.path).observe(duration)

    return response

require_api_key(bootstrap=False)

Return function to validate api key with or without bootstrap.

Source code in opensampl/server/backend/main.py 
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
def require_api_key(bootstrap: bool = False):
    """Return function to validate api key with or without bootstrap."""

    def validate_api_key(api_key: str = Security(api_key_header)) -> str | None:
        """Validate provided API key"""
        if not USE_API_KEY:
            return None  # Security is disabled

        keys = get_keys()
        if not keys and bootstrap:
            logger.warning("No API keys configured; allowing bootstrap API key generation")
            return None
        if api_key not in keys:
            raise HTTPException(status_code=403, detail="Invalid or missing API key")
        return api_key

    return validate_api_key

set_log_level(newloglevel, api_key=Depends(require_api_key()))

Change visible log level in backend container

Source code in opensampl/server/backend/main.py 
161
162
163
164
165
166
@app.get("/setloglevel")
def set_log_level(newloglevel: str, api_key: str = Depends(require_api_key())):
    """Change visible log level in backend container"""
    newloglevel = newloglevel.upper()
    logger.configure(handlers=[{"sink": sys.stderr, "level": newloglevel}])
    return {"loglevel": newloglevel}

write_to_table(payload, api_key=Depends(require_api_key()), session=Depends(get_db))

Write given data to specified table

Source code in opensampl/server/backend/main.py 
180
181
182
183
184
185
186
187
188
189
190
191
192
193
194
195
196
197
198
199
200
201
@app.post("/write_to_table")
def write_to_table(
    payload: WriteTablePayload, api_key: str = Depends(require_api_key()), session: Session = Depends(get_db)
):
    """Write given data to specified table"""
    try:
        load_data.write_to_table(table=payload.table, data=payload.data, if_exists=payload.if_exists, session=session)
        logger.debug(f"Successfully wrote to {payload.table} using: {payload.data}")
        return JSONResponse(content={"message": f"Succeeded loading data into {payload.table}"}, status_code=200)
    except IntegrityError as e:
        if isinstance(e.orig, psycopg2.errors.UniqueViolation):
            return JSONResponse(content={"message": f"Unique violation error: {e}"}, status_code=409)
        return JSONResponse(content={"message": f"Integrity error: {e}"}, status_code=500)
    except SQLAlchemyError as e:
        logger.error(f"SQLAlchemy error: {e}")
        return JSONResponse(content={"message": f"Database error: {e}"}, status_code=500)
    except json.JSONDecodeError as e:
        logger.error(f"JSON decode error: {e}")
        return JSONResponse(content={"message": f"Invalid JSON data: {e}"}, status_code=400)
    except Exception as e:
        logger.error(f"Unexpected error: {e}")
        return JSONResponse(content={"message": f"Failed to load JSON into database: {e}"}, status_code=500)