REST Server

app

FastAPI app factory wrapping BaseLinker.link as a REST endpoint.

The deployer constructs whatever BaseLinker and EntitySource instances they need (model paths, device, checkpoints, API keys, ...) and registers them by name via create_app; the returned app is run the normal way, e.g. uvicorn mymodule:app.

create_app

create_app(
    linkers: dict[str, BaseLinker],
    sources: dict[str, EntitySource] | None = None,
) -> FastAPI

Build a FastAPI app exposing the given linkers/sources over REST.

Parameters:

Name	Type	Description	Default
linkers	dict[str, BaseLinker]	Maps a name (used in the POST /link/{name} URL) to a ready-to-use BaseLinker instance.	required
sources	dict[str, EntitySource] | None	Maps a name to a ready-to-use EntitySource, so it can be referenced by name as a request's dataset2 instead of materializing entities inline. Passed through to link() as-is, preserving search-based blocking behavior.	None

Returns:

Type	Description
FastAPI	A FastAPI app with GET /health, GET /linkers, GET /sources,
FastAPI	and POST /link/{linker_name} routes.

Raises:

Type	Description
OptionalDependencyError	If fastapi isn't installed.

Source code in src/linkingtk/server/app.py

def create_app(
    linkers: dict[str, BaseLinker],
    sources: dict[str, EntitySource] | None = None,
) -> "FastAPI":
    """Build a FastAPI app exposing the given linkers/sources over REST.

    Args:
        linkers: Maps a name (used in the ``POST /link/{name}`` URL) to a
            ready-to-use `BaseLinker` instance.
        sources: Maps a name to a ready-to-use `EntitySource`, so it can be
            referenced by name as a request's `dataset2` instead of
            materializing entities inline. Passed through to `link()`
            as-is, preserving search-based blocking behavior.

    Returns:
        A FastAPI app with `GET /health`, `GET /linkers`, `GET /sources`,
        and `POST /link/{linker_name}` routes.

    Raises:
        OptionalDependencyError: If `fastapi` isn't installed.
    """
    try:
        from fastapi import FastAPI, File, Form, HTTPException, UploadFile
    except ImportError as exc:
        raise OptionalDependencyError("REST server", "server") from exc

    sources = sources or {}
    app = FastAPI(title="linkingtk REST server")

    @app.get("/health")
    def health() -> dict[str, str]:
        return {"status": "ok"}

    @app.get("/linkers")
    def list_linkers() -> list[str]:
        return sorted(linkers)

    @app.get("/sources")
    def list_sources() -> list[str]:
        return sorted(sources)

    @app.post("/link/{linker_name}")
    def link(
        linker_name: str,
        request: str = Form(...),
        graph: UploadFile | None = File(None),  # noqa: B008
        graph_format: str = Form("triples"),
    ) -> list[SchemaAlignmentResult]:
        linker = linkers.get(linker_name)
        if linker is None:
            raise HTTPException(status_code=404, detail=f"Unknown linker {linker_name!r}")

        try:
            body = LinkRequestBody.model_validate_json(request)
        except ValueError as exc:
            raise HTTPException(status_code=400, detail=str(exc)) from exc

        dataset1 = [e.to_entity() for e in body.dataset1]
        dataset2: list[Entity] | EntitySource
        if body.dataset2_entities is not None:
            dataset2 = [e.to_entity() for e in body.dataset2_entities]
        else:
            assert body.dataset2_source is not None  # enforced by LinkRequestBody validator
            source = sources.get(body.dataset2_source)
            if source is None:
                raise HTTPException(
                    status_code=404, detail=f"Unknown source {body.dataset2_source!r}"
                )
            dataset2 = source

        parsed_graph = None
        if graph is not None:
            try:
                parsed_graph = parse_graph(graph.file.read(), graph_format)
            except (ValueError, OptionalDependencyError) as exc:
                raise HTTPException(status_code=400, detail=str(exc)) from exc

        results = linker.link(dataset1, dataset2, parsed_graph)
        return [SchemaAlignmentResult.from_result(r) for r in results]

    return app

schemas

Pydantic request/response models for the REST server.

These mirror the plain dataclasses in linkingtk.core.entity and linkingtk.core.result so the API has a typed, OpenAPI-documented contract, with small converters to/from the real dataclasses used everywhere else in the library.

LinkRequestBody

Bases: BaseModel

The JSON request part of a POST /link/{linker_name} call.

Exactly one of dataset2_entities (an inline target list) or dataset2_source (the name of a pre-registered EntitySource) must be set.

SchemaAlignmentResult

Bases: BaseModel

JSON mirror of AlignmentResult.

SchemaEntity

Bases: BaseModel

JSON mirror of Entity.

graph_io

Parse an uploaded supporting graph into a linkingtk.utils.graph.Graph.

Supports raw inline triples, any RDF serialization rdflib can auto-detect (turtle, N-Triples, RDF/XML, JSON-LD, ...), and NetworkX's GraphML, GEXF, and edge-list file formats. The result is handed to BaseLinker.link as-is -- to_triples normalizes it from there, so this module does no triple conversion of its own.

parse_graph

parse_graph(data: bytes, fmt: str) -> Graph

Parse uploaded graph bytes into a linkingtk.utils.graph.Graph.

Parameters:

Name	Type	Description	Default
data	bytes	The raw uploaded file/body content.	required
fmt	str	One of "triples" (inline JSON list of [s, p, o] triples), "rdf" (any rdflib-parseable serialization), or "graphml"/"gexf"/"edgelist".	required

Raises:

Type	Description
ValueError	If fmt isn't one of the supported values, or the content can't be parsed as that format.

Source code in src/linkingtk/server/graph_io.py

def parse_graph(data: bytes, fmt: str) -> Graph:
    """Parse uploaded graph bytes into a `linkingtk.utils.graph.Graph`.

    Args:
        data: The raw uploaded file/body content.
        fmt: One of ``"triples"`` (inline JSON list of ``[s, p, o]``
            triples), ``"rdf"`` (any rdflib-parseable serialization), or
            ``"graphml"``/``"gexf"``/``"edgelist"``.

    Raises:
        ValueError: If ``fmt`` isn't one of the supported values, or the
            content can't be parsed as that format.
    """
    if fmt == "triples":
        try:
            triples = json.loads(data)
        except json.JSONDecodeError as exc:
            raise ValueError(f"Invalid JSON for graph_format='triples': {exc}") from exc
        return [(s, p, o) for s, p, o in triples]

    if fmt == "rdf":
        try:
            import rdflib
        except ImportError as exc:
            raise OptionalDependencyError("graph_format='rdf'", "graph") from exc

        graph = rdflib.Graph()
        try:
            graph.parse(data=data.decode("utf-8"))
        except Exception as exc:  # rdflib raises many different parser-specific errors
            raise ValueError(f"Could not parse RDF graph: {exc}") from exc
        return graph

    if fmt in _NETWORKX_FORMATS:
        try:
            import networkx as nx
        except ImportError as exc:
            raise OptionalDependencyError(f"graph_format={fmt!r}", "graph") from exc

        reader = {
            "graphml": nx.read_graphml,
            "gexf": nx.read_gexf,
            "edgelist": nx.read_edgelist,
        }[fmt]
        with tempfile.TemporaryDirectory() as tmp_dir:
            tmp_path = Path(tmp_dir) / f"graph.{fmt}"
            tmp_path.write_bytes(data)
            try:
                return reader(tmp_path)
            except Exception as exc:  # networkx raises many different parser-specific errors
                raise ValueError(f"Could not parse {fmt} graph: {exc}") from exc

    raise ValueError(
        f"Unsupported graph_format {fmt!r}; expected one of {sorted(SUPPORTED_FORMATS)}"
    )