Graph Store Protocol commands¶

@app.command(name="upload", help="Upload file(s) to a database")
def upload_command(
    path: Path = typer.Argument(
        ..., help="The path of a file or directory of files to be uploaded."
    ),
    sparql_endpoint: str = typer.Argument(
        ..., help="SPARQL Endpoint URL. E.g. http://localhost:3030/ds"
    ),
    graph_identifier: Annotated[
        str | None,
        typer.Option(
            "--graph",
            "-g",
            help='ID - IRI or URN - of the graph to upload into. If not set, the default graph is targeted. If set to the string "file", the URN urn:file:FILE_NAME will be used per file',
        ),
    ] = None,
    username: Annotated[
        str, typer.Option("--username", "-u", help="Fuseki username.")
    ] = None,
    password: Annotated[
        str, typer.Option("--password", "-p", help="Fuseki password.")
    ] = None,
    timeout: Annotated[
        int, typer.Option("--timeout", "-t", help="Timeout per request")
    ] = 60,
    disable_ssl_verification: Annotated[
        bool,
        typer.Option(
            "--disable-ssl-verification", "-k", help="Disable SSL verification."
        ),
    ] = False,
    host_header: Annotated[
        str | None, typer.Option("--host-header", "-e", help="Override the Host header")
    ] = None,
) -> None:
    """Upload a file or a directory of files with an RDF file extension.

    File extensions: [.nt, .nq, .ttl, .trig, .json, .jsonld, .xml]

    Files are uploaded into their own named graph in the format:
    <urn:file:{file.name}>
    E.g. <urn:file:example.ttl>
    """
    files = []

    if path.is_file():
        files.append(path)
    else:
        files += path.glob("**/*")

    auth = (
        (username, password) if username is not None and password is not None else None
    )

    files = list(filter(lambda f: f.suffix in RDF_SUFFIX_MAP.keys(), files))

    with httpx.Client(
        auth=auth,
        timeout=timeout,
        headers={"Host": host_header} if host_header is not None else {},
        verify=False if disable_ssl_verification else True,
    ) as http_client:
        for file in track(files, description=f"Uploading {len(files)} files..."):
            try:
                if graph_identifier == "file":
                    upload(
                        sparql_endpoint,
                        file,
                        f"urn:file:{file.name}",
                        http_client=http_client,
                    )
                else:
                    upload(
                        sparql_endpoint,
                        file,
                        graph_identifier,
                        http_client=http_client,
                    )  # str and None handled by upload()
            except Exception as err:
                console.print(
                    f"[bold red]ERROR[/bold red] Failed to upload file {file}."
                )
                raise err