Skip to content

ontocast.api.tenancy_resolution

Shared HTTP request tenancy resolution for API routes.

apply_request_tenancy(request, tools, *, active_tenant, active_project, initialize_vector_store) async

Resolve tenant/project and retarget partitioned stores when the client set QS.

Mirrors /process: if tenant or project appears in the query string, resolve with defaults and call :meth:ToolBox.update_tenancy_with_vector_mode when Fuseki/Qdrant partitions are in use. Otherwise return active_* from server startup without retargeting.

Source code in ontocast/api/tenancy_resolution.py 
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
async def apply_request_tenancy(
    request: Request,
    tools: ToolBox,
    *,
    active_tenant: str,
    active_project: str,
    initialize_vector_store: bool,
) -> tuple[str, str]:
    """Resolve tenant/project and retarget partitioned stores when the client set QS.

    Mirrors ``/process``: if ``tenant`` or ``project`` appears in the query string,
    resolve with defaults and call :meth:`ToolBox.update_tenancy_with_vector_mode`
    when Fuseki/Qdrant partitions are in use. Otherwise return ``active_*`` from
    server startup without retargeting.
    """
    if not request_has_tenancy_query_params(request):
        return active_tenant, active_project
    request_tenant = request.query_params.get("tenant", None)
    request_project = request.query_params.get("project", None)
    resolved_tenant, resolved_project = resolve_tenant_project(
        request_tenant, request_project
    )
    if stores_use_tenancy_partitions(tools):
        await tools.update_tenancy_with_vector_mode(
            resolved_tenant,
            resolved_project,
            initialize_vector_store=initialize_vector_store,
            fail_on_vector_store_error=False,
        )
    return resolved_tenant, resolved_project

stores_use_tenancy_partitions(tools)

True when Fuseki and/or Qdrant should be retargeted for tenant/project.

Source code in ontocast/api/tenancy_resolution.py 
10
11
12
13
14
def stores_use_tenancy_partitions(tools: ToolBox) -> bool:
    """True when Fuseki and/or Qdrant should be retargeted for tenant/project."""
    if tools.vector_store is not None:
        return True
    return isinstance(tools.triple_store_manager, FusekiTripleStoreManager)