# retoor <retoor@molodetz.nl>
import inspect
import os
import httpx
from devplacepy.cache import TTLCache
from devplacepy_services.base.db_codec import (
decode_value,
encode_args,
is_write,
is_write_sql,
)
_SERVICE_URL = os.environ.get("DEVPLACE_DB_SERVICE_URL", "http://127.0.0.1:10601").rstrip("/")
_INTERNAL_KEY = os.environ.get("DEVPLACE_GATEWAY_INTERNAL_KEY", "").strip()
_CLIENT: httpx.Client | None = None
# Section 7.3: settings reads tolerate up to 5s staleness. patch_module()
# generically RPCs every devplacepy.database call, bypassing the local
# TTL cache get_setting/get_int_setting had in-process - without this,
# every settings read (rate limiting, maintenance mode, admin dashboards)
# pays a full HTTP round trip to the database broker.
_SETTINGS_CACHE_TTL_SECONDS = 5
_SETTINGS_CACHE = TTLCache(ttl=_SETTINGS_CACHE_TTL_SECONDS, max_size=512)
_CACHED_SETTINGS_FNS = frozenset({"get_setting", "get_int_setting"})
def _service_url() -> str:
return os.environ.get("DEVPLACE_DB_SERVICE_URL", _SERVICE_URL).rstrip("/")
def _headers() -> dict[str, str]:
headers: dict[str, str] = {}
key = os.environ.get("DEVPLACE_GATEWAY_INTERNAL_KEY", _INTERNAL_KEY).strip()
if key:
headers["X-Internal-Key"] = key
return headers
def _client() -> httpx.Client:
global _CLIENT
if _CLIENT is None:
_CLIENT = httpx.Client(timeout=30.0)
return _CLIENT
def _post(path: str, body: dict) -> object:
response = _client().post(
f"{_service_url()}/{path.lstrip('/')}",
json=body,
headers=_headers(),
)
if response.status_code >= 400:
payload = response.json() if response.content else {}
message = payload.get("error", "Database service request failed")
raise RuntimeError(message)
if not response.content:
return None
return decode_value(response.json())
def _invoke_cached(fn_name: str, args, kwargs):
cache_key = f"{fn_name}:{args!r}:{sorted(kwargs.items())!r}"
cached = _SETTINGS_CACHE.get(cache_key)
if cached is not None:
return cached
value = _invoke(fn_name, args, kwargs, write=False)
_SETTINGS_CACHE.set(cache_key, value)
return value
def _invoke(fn_name: str, args, kwargs, *, write: bool = False):
encoded_args, encoded_kwargs = encode_args(args, kwargs)
payload = {
"fn": fn_name,
"args": encoded_args,
"kwargs": encoded_kwargs,
"write": write,
}
result = _post("internal/invoke", payload)
if isinstance(result, dict) and "result" in result:
return result["result"]
return result
class RemoteSearchClause:
def __init__(self, term, fields, author_field=None):
self.term = term.strip()
self.fields = tuple(fields)
self.author_field = author_field
class RemoteUidInClause:
def __init__(self, field, uids):
self.field = field
self.uids = frozenset(uids)
class RemoteTable:
def __init__(self, db: "RemoteDb", name: str) -> None:
self._db = db
self._name = name
self._column_cache = None
def __getattr__(self, name: str):
def caller(*args, **kwargs):
return self._db._table_op(self._name, name, args, kwargs)
return caller
def has_column(self, name: str) -> bool:
cache = self._column_cache
if cache is None:
sample = self.find(_limit=1)
row = next(iter(sample), None)
cache = set(row.keys()) if row else set()
self._column_cache = cache
return name in cache
def count(self, **kwargs):
return self._db._table_op(self._name, "count", [], kwargs)
@property
def table(self):
return self
@property
def exists(self) -> bool:
return self._name in self._db.tables
class RemoteDb:
def __init__(self) -> None:
self._tables_cache: list[str] | None = None
@property
def tables(self) -> list[str]:
if self._tables_cache is None:
result = _post("internal/db-op", {"op": "tables"})
self._tables_cache = list(result or [])
return self._tables_cache
def __getitem__(self, name: str) -> RemoteTable:
return RemoteTable(self, name)
def query(self, sql: str, **params):
encoded_args, encoded_kwargs = encode_args((sql,), params)
result = _post(
"internal/db-op",
{
"op": "query",
"args": encoded_args,
"kwargs": encoded_kwargs,
"write": is_write_sql(sql),
},
)
return result or []
def _table_op(self, table: str, method: str, args, kwargs, *, write: bool = False):
encoded_args, encoded_kwargs = encode_args(args, kwargs)
result = _post(
"internal/db-op",
{
"op": "table_op",
"table": table,
"method": method,
"args": encoded_args,
"kwargs": encoded_kwargs,
"write": write,
},
)
if method in {"insert", "update", "delete"}:
self._tables_cache = None
return result
@property
def executable(self):
return self
@property
def in_transaction(self) -> bool:
return False
def __enter__(self):
return self
def __exit__(self, exc_type, exc, tb):
return False
_LOCAL_REMOTE = frozenset(
{
"get_table",
"refresh_snapshot",
"_in_clause",
"_now_iso",
"text_search_clause",
}
)
def _remote_text_search_clause(
table, search, fields=("title", "description"), author_field=None
):
term = (search or "").strip()
if not term:
return None
if type(table).__name__ == "RemoteTable":
return RemoteSearchClause(term, fields, author_field)
from devplacepy.database.content import text_search_clause as local_clause
return local_clause(table, search, fields, author_field=author_field)
def _remote_get_table(name: str):
import devplacepy.database.core as core
return core.db[name]
def _remote_refresh_snapshot() -> None:
return None
def patch_module(module) -> None:
import devplacepy.database as db_module
for name in db_module.__all__:
if name in _LOCAL_REMOTE:
continue
target = getattr(module, name, None)
if target is None or not callable(target):
continue
if inspect.isclass(target):
continue
def make_wrapper(fn_name: str, fn_write: bool):
if fn_name in _CACHED_SETTINGS_FNS:
def wrapper(*args, **kwargs):
return _invoke_cached(fn_name, args, kwargs)
wrapper.__name__ = fn_name
return wrapper
def wrapper(*args, **kwargs):
return _invoke(fn_name, args, kwargs, write=fn_write)
wrapper.__name__ = fn_name
return wrapper
setattr(module, name, make_wrapper(name, is_write(name)))
def activate() -> None:
import devplacepy.database.core as core
core.db = RemoteDb()
import devplacepy.database as db_module
patch_module(db_module)
for submodule_name in (
"settings",
"users",
"relations",
"pagination",
"soft_delete",
"engagement",
"usage",
"awards",
"seo_meta",
"activity",
"customization",
"email",
"notifications",
"forks",
"follows",
"deepsearch",
"ranking",
"comments",
"content",
"attachments_data",
"stats",
"schema",
):
try:
submodule = __import__(
f"devplacepy.database.{submodule_name}",
fromlist=[submodule_name],
)
except ImportError:
continue
patch_module(submodule)
for external_name in (
"devplacepy.services.statistics.tracking",
"devplacepy.services.base",
"devplacepy.attachments",
"devplacepy.project_files",
):
try:
external = __import__(external_name, fromlist=[external_name.split(".")[-1]])
except ImportError:
continue
if hasattr(external, "db"):
external.db = RemoteDb()
db_module.db = core.db
db_module.get_table = _remote_get_table
core.get_table = _remote_get_table
db_module.refresh_snapshot = _remote_refresh_snapshot
core.refresh_snapshot = _remote_refresh_snapshot
db_module.text_search_clause = _remote_text_search_clause
import devplacepy.database.content as content_module
content_module.text_search_clause = _remote_text_search_clause
for submodule_name in (
"settings",
"users",
"relations",
"pagination",
"soft_delete",
"engagement",
"usage",
"awards",
"seo_meta",
"activity",
"customization",
"email",
"notifications",
"forks",
"follows",
"deepsearch",
"ranking",
"comments",
"content",
"attachments_data",
"stats",
"schema",
):
try:
submodule = __import__(
f"devplacepy.database.{submodule_name}",
fromlist=[submodule_name],
)
except ImportError:
continue
if hasattr(submodule, "db"):
submodule.db = core.db
Reference in New Issue View Git Blame Copy Permalink