@app.command(name="sparql", help="SPARQL queries to local RDF files or a database")
def sparql_command(
path_or_url: Path,
q: str,
response_format: str = typer.Option(
"table",
"--response-format",
"-f",
help="The response format of the SPARQL query. Either 'table' (default), 'json' or 'csv'",
),
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:
"""SPARQL queries a local file or SPARQL Endpoint"""
if str(path_or_url).startswith("http"):
path_or_url = str(path_or_url).replace(":/", "://")
if isinstance(q, str):
if len(q) < 260:
if Path(q).is_file():
q = Path(q).read_text()
if response_format not in ["table", "json", "csv"]:
raise typer.BadParameter(
"response_format must be either 'table' (default), 'json' or 'csv'"
)
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:
r = query(path_or_url, q, http_client=http_client, return_format="python")
if r == "":
console.print("Operation completed successfully")
return
# if it is a graph, just print return the serialized form plainly, not via console.print()
# to avoid terminal width breaking long literals, as per Issue 37
if isinstance(r, rdflib.Graph):
print(r.serialize(format="longturtle"))
elif response_format == "table":
console.print(format_sparql_response_as_rich_table(r, q))
elif response_format == "csv":
console.print(format_sparql_response_as_csv(r, q))
else:
print(format_sparql_response_as_json(r))