retoor
d31ccba6c1
Add a new `/dbapi` router package providing a generic database API over `dataset`, restricted to admin sessions, admin API keys, and the internal gateway key. Includes: - `tables.py`: list all tables and inspect table schemas - `crud.py`: full CRUD operations (GET, POST, PATCH, DELETE) with soft-delete awareness, born-live inserts, `?include_deleted`, `.../restore`, and `?hard=true` purge - `query.py`: validated read-only SELECT execution via sqlglot parsing, classification, and EXPLAIN dry-run; async query jobs with WebSocket streaming via `DbApiJobService` - `nl.py`: natural-language-to-SQL conversion using the platform AI gateway with re-prompting until validation passes Also register `DbApiJobService` and `PubSubService` in the service manager, add `DBAPI_DIR` to config data paths, and force cleartext `http://` connections to HTTP/1.1 in `curl_transport` to fix large request failures against uvicorn's HTTP/1.1-only internal gateway.
59 lines
1.6 KiB
Python
59 lines
1.6 KiB
Python
# retoor <retoor@molodetz.nl>
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
import pytest
|
|
from fastapi.testclient import TestClient
|
|
|
|
from devplacepy.database import get_table, internal_gateway_key, purge, refresh_snapshot
|
|
from devplacepy.main import app
|
|
from devplacepy.services.manager import service_manager
|
|
from devplacepy.utils import generate_uid
|
|
|
|
|
|
def _seed_schema():
|
|
now = datetime.now(timezone.utc).isoformat()
|
|
users = get_table("users")
|
|
if "username" not in users.columns:
|
|
users.insert(
|
|
{
|
|
"uid": generate_uid(),
|
|
"username": "dbapi_seed",
|
|
"email": "seed@dbapi.test",
|
|
"role": "Member",
|
|
"xp": 0,
|
|
"level": 1,
|
|
"created_at": now,
|
|
}
|
|
)
|
|
bookmarks = get_table("bookmarks")
|
|
if "user_uid" not in bookmarks.columns:
|
|
seed = generate_uid()
|
|
bookmarks.insert(
|
|
{
|
|
"uid": seed,
|
|
"user_uid": "_seedcols",
|
|
"target_type": "post",
|
|
"target_uid": "_s",
|
|
"created_at": now,
|
|
"deleted_at": None,
|
|
"deleted_by": None,
|
|
}
|
|
)
|
|
purge("bookmarks", uid=seed)
|
|
refresh_snapshot()
|
|
|
|
|
|
@pytest.fixture(scope="module")
|
|
def client():
|
|
with TestClient(app) as test_client:
|
|
service_manager.set_lock_owner(True)
|
|
_seed_schema()
|
|
yield test_client
|
|
service_manager.set_lock_owner(False)
|
|
|
|
|
|
@pytest.fixture
|
|
def auth(client):
|
|
return {"X-API-KEY": internal_gateway_key()}
|