FastAPI (optional extra)

Requires pip install "sparqlmodel[fastapi]".

FastAPI helpers (optional sparqlmodel[fastapi] extra).

sparqlmodel.fastapi.async_http_store_lifespan(app, endpoint, *, auth=None, headers=None, prefixes=None, **http_kwargs)[source]

AsyncHttpStore lifespan: init_async_app() on startup, aclose on shutdown.

Return type:: AsyncIterator[None]

sparqlmodel.fastapi.async_session_dependency(store=None, *, store_factory=None, prefixes=None, autoflush=True, rollback_on_error=True, close_on_exit=None)[source]

Build a custom async get_session for dependency overrides.

Return type:: Callable[[Request], AsyncIterator[AsyncSPARQLSession]]

async sparqlmodel.fastapi.get_async_session(request)[source]

Yield one AsyncSPARQLSession per request.

Return type:: AsyncIterator[AsyncSPARQLSession]

sparqlmodel.fastapi.get_session(request)[source]

Yield one SPARQLSession per request (use with Depends).

Mirrors SQLModel / SQLAlchemy:

def get_session():
    with Session(engine) as session:
        yield session

The shared store comes from init_app() or http_store_lifespan(). When neither is configured, each request gets an isolated in-memory store.

Return type:: Generator[SPARQLSession, None, None]

sparqlmodel.fastapi.http_store_lifespan(app, endpoint, *, auth=None, headers=None, prefixes=None, **http_kwargs)[source]

HttpStore lifespan: init_app() on startup, close the store on shutdown.

Return type:: AsyncIterator[None]

sparqlmodel.fastapi.init_app(app, store, *, prefixes=None, autoflush=True, rollback_on_error=True)[source]

Attach a shared store and session options to app.state (like a SQLAlchemy engine).

sparqlmodel.fastapi.init_async_app(app, store, *, prefixes=None, autoflush=True, rollback_on_error=True)[source]

Attach a shared async store and session options to app.state.

sparqlmodel.fastapi.jsonld_response(model, *, status_code=200)[source]

Return a JSON-LD HTTP response for a model or graph.

Return type:: Response

sparqlmodel.fastapi.negotiated_response(request, model, *, formats=None)[source]

Return Turtle or JSON-LD based on Accept (defaults to Turtle).

Return type:: Response

sparqlmodel.fastapi.session_dependency(store=None, *, store_factory=None, prefixes=None, autoflush=True, rollback_on_error=True, close_on_exit=None)[source]

Build a custom get_session for app.dependency_overrides or multi-store apps.

Return type:: Callable[[Request], Generator[SPARQLSession, None, None]]

sparqlmodel.fastapi.turtle_response(model, *, status_code=200)[source]

Return a Turtle HTTP response for a model or graph.

Return type:: Response

FastAPI session dependency (SQLModel / SQLAlchemy style).

sparqlmodel.fastapi.deps.init_app(app, store, *, prefixes=None, autoflush=True, rollback_on_error=True)[source]

Attach a shared store and session options to app.state (like a SQLAlchemy engine).

sparqlmodel.fastapi.deps.get_session(request)[source]

Yield one SPARQLSession per request (use with Depends).

Mirrors SQLModel / SQLAlchemy:

def get_session():
    with Session(engine) as session:
        yield session

The shared store comes from init_app() or http_store_lifespan(). When neither is configured, each request gets an isolated in-memory store.

Return type:: Generator[SPARQLSession, None, None]

sparqlmodel.fastapi.deps.http_store_lifespan(app, endpoint, *, auth=None, headers=None, prefixes=None, **http_kwargs)[source]

HttpStore lifespan: init_app() on startup, close the store on shutdown.

Return type:: AsyncIterator[None]

sparqlmodel.fastapi.deps.session_dependency(store=None, *, store_factory=None, prefixes=None, autoflush=True, rollback_on_error=True, close_on_exit=None)[source]

Build a custom get_session for app.dependency_overrides or multi-store apps.

Return type:: Callable[[Request], Generator[SPARQLSession, None, None]]

sparqlmodel.fastapi.deps.init_async_app(app, store, *, prefixes=None, autoflush=True, rollback_on_error=True)[source]

Attach a shared async store and session options to app.state.

async sparqlmodel.fastapi.deps.get_async_session(request)[source]

Yield one AsyncSPARQLSession per request.

Return type:: AsyncIterator[AsyncSPARQLSession]

sparqlmodel.fastapi.deps.async_http_store_lifespan(app, endpoint, *, auth=None, headers=None, prefixes=None, **http_kwargs)[source]

AsyncHttpStore lifespan: init_async_app() on startup, aclose on shutdown.

Return type:: AsyncIterator[None]

sparqlmodel.fastapi.deps.async_session_dependency(store=None, *, store_factory=None, prefixes=None, autoflush=True, rollback_on_error=True, close_on_exit=None)[source]

Build a custom async get_session for dependency overrides.

Return type:: Callable[[Request], AsyncIterator[AsyncSPARQLSession]]