trustgraph/trustgraph-cli/scripts/tg-graph-to-turtle

#!/usr/bin/env python3

"""
Connects to the graph query service and dumps all graph edges in Turtle
format.
"""

import rdflib
import io
import sys
import argparse
import os

from trustgraph.api import Api, Uri

default_url = os.getenv("TRUSTGRAPH_URL", 'http://localhost:8088/')
default_user = 'trustgraph'
default_collection = 'default'

def show_graph(url, flow_id, user, collection):

    api = Api(url).flow().id(flow_id)

    rows = api.triples_query(
        s=None, p=None, o=None,
        user=user, collection=collection,
        limit=10_000)

    g = rdflib.Graph()

    for row in rows:

        sv = rdflib.term.URIRef(row.s)
        pv = rdflib.term.URIRef(row.p)

        if isinstance(row.o, Uri):

            # Skip malformed URLs with spaces in
            if " " in row.o:
                continue

            ov = rdflib.term.URIRef(row.o)

        else:

            ov = rdflib.term.Literal(row.o)

        g.add((sv, pv, ov))

    g.serialize(destination="output.ttl", format="turtle")

    buf = io.BytesIO()

    g.serialize(destination=buf, format="turtle")

    sys.stdout.write(buf.getvalue().decode("utf-8"))


def main():

    parser = argparse.ArgumentParser(
        prog='tg-graph-to-turtle',
        description=__doc__,
    )

    parser.add_argument(
        '-u', '--api-url',
        default=default_url,
        help=f'API URL (default: {default_url})',
    )

    parser.add_argument(
        '-f', '--flow-id',
        default="0000",
        help=f'Flow ID (default: 0000)'
    )

    parser.add_argument(
        '-U', '--user',
        default=default_user,
        help=f'User ID (default: {default_user})'
    )

    parser.add_argument(
        '-C', '--collection',
        default=default_collection,
        help=f'Collection ID (default: {default_collection})'
    )

    args = parser.parse_args()

    try:

        show_graph(
            url = args.api_url,
            flow_id = args.flow_id,
            user = args.user,
            collection = args.collection,
        )

    except Exception as e:

        print("Exception:", e, flush=True)

main()