wehub-resource-sync
fed8b2eed7
Build and push multi-arch DocsGPT Docker image / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Backend release / release (push) Has been cancelled
Bandit Security Scan / bandit_scan (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push multi-arch DocsGPT Docker image / manifest (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/amd64, ubuntu-latest, amd64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / build (linux/arm64, ubuntu-24.04-arm, arm64) (push) Has been cancelled
Build and push DocsGPT FE Docker image for development / manifest (push) Has been cancelled
Python linting / ruff (push) Has been cancelled
Run python tests with pytest / Run tests and count coverage (3.12) (push) Has been cancelled
React Widget Build / build (push) Has been cancelled
61 lines
2.0 KiB
Python
61 lines
2.0 KiB
Python
"""Repository for the ``app_metadata`` singleton key/value table.
|
|
|
|
Owns the instance-wide state the version-check client needs:
|
|
``instance_id`` (anonymous UUID sent with each check) and
|
|
``version_check_notice_shown`` (one-shot flag for the first-run
|
|
telemetry notice). Kept deliberately generic so future one-off config
|
|
values can piggyback without a new migration each time.
|
|
"""
|
|
|
|
from __future__ import annotations
|
|
|
|
import uuid
|
|
from typing import Optional
|
|
|
|
from sqlalchemy import Connection, text
|
|
|
|
|
|
class AppMetadataRepository:
|
|
"""Postgres-backed ``app_metadata`` store. Tiny by design."""
|
|
|
|
def __init__(self, conn: Connection) -> None:
|
|
self._conn = conn
|
|
|
|
def get(self, key: str) -> Optional[str]:
|
|
row = self._conn.execute(
|
|
text("SELECT value FROM app_metadata WHERE key = :key"),
|
|
{"key": key},
|
|
).fetchone()
|
|
return row[0] if row is not None else None
|
|
|
|
def set(self, key: str, value: str) -> None:
|
|
self._conn.execute(
|
|
text(
|
|
"INSERT INTO app_metadata (key, value) VALUES (:key, :value) "
|
|
"ON CONFLICT (key) DO UPDATE SET value = EXCLUDED.value"
|
|
),
|
|
{"key": key, "value": value},
|
|
)
|
|
|
|
def get_or_create_instance_id(self) -> str:
|
|
"""Return the anonymous instance UUID, generating one if absent.
|
|
|
|
Uses ``INSERT ... ON CONFLICT DO NOTHING`` + re-read so two
|
|
workers racing on the very first startup converge on a single
|
|
UUID instead of each persisting their own.
|
|
"""
|
|
existing = self.get("instance_id")
|
|
if existing:
|
|
return existing
|
|
candidate = str(uuid.uuid4())
|
|
self._conn.execute(
|
|
text(
|
|
"INSERT INTO app_metadata (key, value) VALUES ('instance_id', :value) "
|
|
"ON CONFLICT (key) DO NOTHING"
|
|
),
|
|
{"value": candidate},
|
|
)
|
|
# Re-read: if another worker won the race, their UUID is now authoritative.
|
|
winner = self.get("instance_id")
|
|
return winner or candidate
|