Fuseki commands¶

@app.command(name="ping", help="Check if the server is alive")
def ping_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    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,
) -> None:
    """Check if the Fuseki server is alive.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(ping(fuseki_url, http_client=http_client))
        except Exception as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to ping server at {fuseki_url}."
            )
            raise err

@app.command(name="server", help="Get basic server info")
def server_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    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,
) -> None:
    """Get basic Fuseki server info.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(server(fuseki_url, http_client=http_client))
        except Exception as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to get server info at {fuseki_url}."
            )
            raise err

@app.command(name="stats", help="Request statistics for all datasets or for one")
def stats_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    dataset_name: Annotated[
        str,
        typer.Option("--dataset-name", "-d", help="The dataset to get statistics for"),
    ] = 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,
) -> None:
    """Request statistics for all datasets, or one named dataset, on the Fuseki server.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        dataset_name: The dataset to get statistics for. If not given, statistics for all datasets are returned.
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(stats(fuseki_url, dataset_name, http_client=http_client))
        except Exception as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to get stats at {fuseki_url}."
            )
            raise err

def backup_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    dataset_name: str = typer.Argument(..., help="The name of the dataset to back up"),
    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,
) -> None:
    """Ask the Fuseki server to create a backup of a dataset.

    NOTE: The associated library function is not yet implemented, so this command is not currently available through the CLI.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        dataset_name: The name of the dataset to back up.
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(backup(fuseki_url, dataset_name, http_client=http_client))
        except Exception as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to create backup of dataset {dataset_name} at {fuseki_url}."
            )
            raise err

def backups_list_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    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,
) -> None:
    """List all existing backups on the Fuseki server.

    NOTE: The related backup library function is not yet implemented, so this command is not currently available through the CLI.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(backups_list(fuseki_url, http_client=http_client))
        except Exception as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to list backups at {fuseki_url}."
            )
            raise err

def sleep_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    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,
) -> None:
    """Tell the Fuseki server to sleep.

    NOTE: The associated library function is not yet implemented, so this command is not currently available through the CLI.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(sleep(fuseki_url, http_client=http_client))
        except Exception as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to put server to sleep at {fuseki_url}."
            )
            raise err

@app.command(name="tasks", help="List running tasks or get one")
def tasks_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    task_name: Annotated[
        str,
        typer.Option("--task-name", "-n", help="The task to get"),
    ] = 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,
) -> None:
    """List tasks currently running on the Fuseki server, or get one named task.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        task_name: The task to get. If not given, all running tasks are listed.
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(tasks(fuseki_url, task_name, http_client=http_client))
        except Exception as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to list tasks at {fuseki_url}."
            )
            raise err

@app.command(name="metrics", help="Get server metrics")
def metrics_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    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,
) -> None:
    """Get metrics for the Fuseki server.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(metrics(fuseki_url, http_client=http_client))
        except Exception as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to get metrics at {fuseki_url}."
            )
            raise err

@app.command(name="describe", help="Get the list of datasets or describe one")
def describe_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    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,
) -> None:
    """Get the list of datasets on the Fuseki server, or describe one.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(describe(fuseki_url, http_client=http_client))
        except Exception as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to describe datasets at {fuseki_url}."
            )
            raise err

@app.command(name="create", help="Create a new dataset")
def create_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    dataset_name: str | None = typer.Argument(
        None, help="The name of the dataset to create"
    ),
    dataset_type: str = typer.Option(
        "tdb2", help=f"Dataset type. Options: {dataset_type_options}"
    ),
    config: Path | None = typer.Option(None, help="Assembler file"),
    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,
) -> None:
    """Create a new dataset on the Fuseki server.

    Either `dataset_name` or `config` must be given, not both.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        dataset_name: The name of the dataset to create, using `dataset_type`.
        dataset_type: The dataset type to create, when `dataset_name` is given. Options: `mem`, `tdb`, `tdb1`, `tdb2`.
        config: An assembler file describing the dataset to create, as an alternative to `dataset_name`.
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        typer.BadParameter: If both `dataset_name` and `config` are given, if neither is given, or if `dataset_type` is not a valid option.
        Exception: If the request to the Fuseki server fails.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    if dataset_name and config:
        raise typer.BadParameter("Only dataset name or --config is allowed, not both.")

    if dataset_name and dataset_type not in dataset_type_options:
        raise typer.BadParameter(
            f"Invalid dataset type '{dataset_type}'. Options: {dataset_type_options}"
        )

    if dataset_name:
        with httpx.Client(auth=auth, timeout=timeout) as http_client:
            try:
                console.print(
                    create(
                        fuseki_url, dataset_name, dataset_type, http_client=http_client
                    )
                )
            except Exception as err:
                console.print(
                    f"[bold red]ERROR[/bold red] Failed to create dataset {dataset_name} at {fuseki_url}."
                )
                raise err
    else:
        if config is None:
            raise typer.BadParameter(
                "Either dataset name or assembler config file must be provided."
            )
        with httpx.Client(auth=auth, timeout=timeout) as http_client:
            try:
                console.print(create(fuseki_url, config, http_client=http_client))
            except Exception as err:
                console.print(
                    f"[bold red]ERROR[/bold red] Failed to create dataset {dataset_name} at {fuseki_url}."
                )
                raise err

@app.command(name="delete", help="Delete a dataset")
def delete_command(
    fuseki_url: str = typer.Argument(
        ..., help="Fuseki base URL. E.g. http://localhost:3030"
    ),
    dataset_name: str = typer.Argument(..., help="The name of the dataset to delete"),
    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,
):
    """Delete a dataset from the Fuseki server.

    Args:
        fuseki_url: Fuseki base URL. E.g. http://localhost:3030
        dataset_name: The name of the dataset to delete.
        username: Fuseki username, if the server requires authentication.
        password: Fuseki password, if the server requires authentication.
        timeout: Timeout per request, in seconds.

    Raises:
        FusekiError: If the dataset fails to delete.
    """
    auth = (
        (username, password) if username is not None and password is not None else None
    )

    with httpx.Client(auth=auth, timeout=timeout) as http_client:
        try:
            console.print(delete(fuseki_url, dataset_name, http_client=http_client))
        except FusekiError as err:
            console.print(
                f"[bold red]ERROR[/bold red] Failed to delete dataset {dataset_name} at {fuseki_url}."
            )
            raise err