sparqlquery

A commandline tool for querying with SPARQL on local files and remote sparql endpoints with custom serialization.

example usage:

    sq path/to/data.ttl -q "SELECT ?x WHERE {?x a foaf:Person. }"
    rdfpipe test.ttl | sparqlquery - -q "SELECT ?x WHERE {?x a foaf:Person. }" --format json
    sq data.ttl -q "ASK {:john a foaf:Person}" --format xml | grep true
    sq path/to/data.ttl --query-file query.rq
    sq data1.ttl data2.ttl -q "DESCRIBE <http://example.com/john>" --format turtle:+spacious
    sq http://example.com/sparqlendpoint --query-file query.rq
    sq http://example.com/sparqlendpoint --query-file query.rq --username user --password secret
    sq /pyth/to/berkeley.db -q "SELECT ?x WHERE {?x a foaf:Person. }" --remote-storetype BerkeleyDB

Tip: You can check the truth value for an ASK query, by regex in stdout for ‘true’ or ‘false’.

Functions:

• sparqlquery –

__all__ module-attribute

__all__ = ['sparqlquery']

InvalidQueryError

Bases: Exception

main

main()

Source code in rdflib/tools/sparqlquery.py

def main():
    try:
        (
            endpoints,
            remote_storetype,
            warn,
            opts,
        ) = parse_args()
    except _PrintHelpError:
        exit()
    except (_ArgumentError, argparse.ArgumentError) as err:
        print(err, file=sys.stderr)
        exit(2)

    if warn:
        loglevel = logging.WARNING
    else:
        loglevel = logging.CRITICAL
    logging.basicConfig(level=loglevel)

    sparqlquery(
        endpoints,
        remote_storetype=remote_storetype,
        **opts,
    )

parse_args

parse_args()

Source code in rdflib/tools/sparqlquery.py

def parse_args():
    extra_kwargs: Dict[str, Any] = {}
    if sys.version_info > (3, 9):
        extra_kwargs["exit_on_error"] = False
    parser = argparse.ArgumentParser(
        prog="sparqlquery",
        description=__doc__,
        add_help=False,  # add dynamic epilog before help is added
        formatter_class=argparse.RawDescriptionHelpFormatter,
        # else __doc__ wont be printed on error:
        **extra_kwargs,
    )
    parser.add_argument(
        "-q",
        "--query",
        type=str,
        help="Sparql query. Cannot be set together with -qf/--queryfile.",
    )
    parser.add_argument(
        "-qf",
        "--queryfile",
        type=str,
        help="File from where the sparql query is read. "
        "Cannot be set together with -q/--query",
    )
    parser.add_argument(
        "-f",
        "--format",
        type=str,
        help="Print sparql result in given format. "
        "Defaults to 'json' on SELECT, to 'xml' on ASK "
        "and to 'turtle' on DESCRIBE and CONSTRUCT. "
        "Keywords as described in epilog can be given "
        "after format like: "
        "FORMAT:(+)KW1,-KW2,KW3=VALUE.",
    )
    opts: Dict[str, Any]
    opts, parser.epilog = _extract_query_and_format(parser)

    parser.add_argument(
        "endpoint",
        nargs="+",
        type=str,
        help="Endpoints for sparql query. "
        "Can be set to multiple files. "
        "Reads from stdin if '-' is given. ",
    )
    parser.add_argument(
        "-w",
        "--warn",
        action="store_true",
        default=False,
        help="Output warnings to stderr " "(by default only critical errors).",
    )
    parser.add_argument(
        "-h",
        "--help",
        # action="store_true",
        # default=False,
        action="help",
        help="show help message and exit. "
        "Also prints information about given format.",
    )
    parser.add_argument(
        "-u", "--username", type=str, help="Username used during authentication."
    )
    parser.add_argument(
        "-p", "--password", type=str, help="Password used during authentication."
    )
    parser.add_argument(
        "-rs",
        "--remote-storetype",
        type=str,
        help="You can specify which storetype should be used. "
        "Can only be set, when using a single endpoint and not stdin. "
        "Will default to 'SparqlStore' when endpoint is an internetaddress.",
    )

    try:  # catch error because exit_on_error=False
        args = parser.parse_args()
    except argparse.ArgumentError as err:
        parser.print_help()
        raise err

    forbidden_format_keywords = [
        x
        for x in opts.get("result_keywords", dict())
        if x in {"self", "stream", "encoding", "format"}
    ]
    if forbidden_format_keywords:
        raise _ArgumentError(
            "'self', 'stream', 'encoding' and 'format' "
            "mustnt be used as keywords for format."
        )

    if opts.get("query") is None:
        parser.print_help()
        raise _ArgumentError("Either -q/--query or -qf/--queryfile must be provided")

    remote_storetype = args.remote_storetype

    if len(args.endpoint) == 1:
        if args.endpoint[0] == "-":
            if remote_storetype is not None:
                raise _ArgumentError(
                    "Cant us remote graphtype, when endpoint is stdin(-)"
                )
            endpoints = []
            opts["use_stdin"] = True
        elif _dest_is_internet_addr(args.endpoint[0]):
            endpoints = args.endpoint
            if remote_storetype is None:
                remote_storetype = "SPARQLStore"
        else:
            endpoints = args.endpoint
    else:
        if remote_storetype is not None:
            raise _ArgumentError(
                "If remote graphtype is set, only a single endpoint is valid."
            )
        endpoints = list(args.endpoint)
        if any(not (_dest_is_local(x)) for x in args.endpoint):
            raise NotImplementedError(
                "If multiple endpoints are given, all must be local files."
            )

    if args.username is not None and args.password is not None:
        if remote_storetype not in ["SPARQLStore"]:
            raise _ArgumentError(
                "Can use password and username only, "
                "when remote-storetype is 'SPARQLStore'."
            )
        opts["auth"] = (args.username, args.password)
    elif args.username is None and args.password is None:
        pass
    else:
        parser.print_help()
        raise _ArgumentError("User only provided one of password and username")

    return endpoints, remote_storetype, args.warn, opts

sparqlquery

sparqlquery(endpoints: List[str], query: str, result_format: Optional[str] = None, result_keywords: Dict[str, str] = {}, auth: Optional[Tuple[str, str]] = None, use_stdin: bool = False, remote_storetype: Optional[str] = None)

Source code in rdflib/tools/sparqlquery.py

def sparqlquery(
    endpoints: List[str],
    query: str,
    result_format: Optional[str] = None,
    result_keywords: Dict[str, str] = {},
    auth: Optional[Tuple[str, str]] = None,
    use_stdin: bool = False,
    remote_storetype: Optional[str] = None,
):
    if use_stdin:
        g = Graph().parse(sys.stdin)
    else:
        g = _get_graph(endpoints, auth, remote_storetype)
    try:
        results: Result = g.query(query)
    except ParseException as err:
        raise InvalidQueryError(query) from err

    if result_format is not None:
        ret_bytes = results.serialize(format=result_format, **result_keywords)
    else:
        ret_bytes = results.serialize(**result_keywords)
    if ret_bytes is not None:
        print(ret_bytes.decode())