Spaces:
Sleeping
Sleeping
| """ | |
| The single seam between the org/role model (and everything else that stores operational | |
| metadata) and its actual backing store. Every caller uses get_doc/set_doc/update_doc/ | |
| delete_doc/query_prefix from THIS module, never backend.local_store or the real Firestore | |
| client directly. | |
| LOCAL_MODE governs which backing store this module uses — real Firestore (a real GCP | |
| project + service-account credential now exist, see backend/firebase_app.py) when false, | |
| the local SQLite shim when true (the default in tests — see tests/conftest.py, which forces | |
| LOCAL_MODE=true regardless of .env so the suite never touches real Firestore). | |
| Firestore's own Python client accepts the same slash-delimited path strings our path scheme | |
| already uses (`db.document("orgs/abc123")`, `db.collection("orgs/abc123/mailboxes")`) — no | |
| translation layer needed between our path convention and Firestore's. | |
| """ | |
| from __future__ import annotations | |
| from firebase_admin import firestore | |
| from backend import local_store | |
| from backend.firebase_app import get_app | |
| from backend.local_mode import LOCAL_MODE | |
| def _db(): | |
| get_app() | |
| return firestore.client() | |
| def get_doc(path: str) -> dict | None: | |
| if LOCAL_MODE: | |
| return local_store.get_doc(path) | |
| snap = _db().document(path).get() | |
| return snap.to_dict() if snap.exists else None | |
| def set_doc(path: str, data: dict) -> None: | |
| if LOCAL_MODE: | |
| local_store.set_doc(path, data) | |
| return | |
| _db().document(path).set(data) | |
| def update_doc(path: str, partial: dict) -> dict: | |
| """Shared across both backends: read-modify-write via this module's own get_doc/ | |
| set_doc, so the merge semantics (raise if the doc doesn't exist) stay identical | |
| regardless of which store is active — no need to duplicate the logic per backend.""" | |
| existing = get_doc(path) | |
| if existing is None: | |
| raise KeyError(f"No document at {path!r} to update") | |
| existing.update(partial) | |
| set_doc(path, existing) | |
| return existing | |
| def delete_doc(path: str) -> None: | |
| if LOCAL_MODE: | |
| local_store.delete_doc(path) | |
| return | |
| _db().document(path).delete() | |
| def query_prefix(prefix: str) -> dict[str, dict]: | |
| """All documents under a collection path (e.g. "orgs/abc123/mailboxes/"), keyed by | |
| their full path — matches local_store.query_prefix's contract exactly.""" | |
| if LOCAL_MODE: | |
| return local_store.query_prefix(prefix) | |
| collection_path = prefix.rstrip("/") | |
| docs = _db().collection(collection_path).stream() | |
| return {f"{collection_path}/{doc.id}": doc.to_dict() for doc in docs} | |