ODRL Access Grant Nanopublication Generator

Creates access grant nanopublications that record when a DID is granted access to a dataset. These serve as an immutable audit trail — each grant is a signed, timestamped record.

In production, these are published automatically by GitHub Actions when an access request is approved. This notebook is for manual grants or testing.

📝 SECTION 1: INPUT FILE (EDIT THIS)¶

CONFIG_FILE = "config/hamburg_access_grant.json"

OUTPUT_DIR = "output/"

⚙️ SECTION 2: SETUP¶

import json
from pathlib import Path
from datetime import datetime, timezone
from rdflib import Dataset, Namespace, URIRef, Literal, BNode
from rdflib.namespace import RDF, RDFS, XSD, FOAF

NP = Namespace("http://www.nanopub.org/nschema#")
DCT = Namespace("http://purl.org/dc/terms/")
NT = Namespace("https://w3id.org/np/o/ntemplate/")
NPX = Namespace("http://purl.org/nanopub/x/")
PROV = Namespace("http://www.w3.org/ns/prov#")
ORCID = Namespace("https://orcid.org/")
ODRL = Namespace("http://www.w3.org/ns/odrl/2/")
FAIR = Namespace("https://fair2adapt.eu/ns/")

# Template URIs
ODRL_GRANT_TEMPLATE = URIRef("https://w3id.org/np/RAd1X8liGs-fjmbkrN514sL3CueyOVOjX3Bax63br6HYM")
PROV_TEMPLATE = URIRef("https://w3id.org/np/RA7lSq6MuK_TIC6JMSHvLtee3lpLoZDOqLJCLXevnrPoU")
PUBINFO_TEMPLATE_1 = URIRef("https://w3id.org/np/RA0J4vUn_dekg-U1kK3AOEt02p9mT2WO03uGxLDec1jLw")
PUBINFO_TEMPLATE_2 = URIRef("https://w3id.org/np/RAukAcWHRDlkqxk7H2XNSegc1WnHI569INvNr-xdptDGI")

ODRL_ACTIONS = {
    "use": ODRL.use,
    "reproduce": ODRL.reproduce,
    "distribute": ODRL.distribute,
}

print("✓ Setup complete")

📖 SECTION 3: LOAD & VALIDATE¶

print(f"Loading: {CONFIG_FILE}")

with open(CONFIG_FILE, 'r', encoding='utf-8') as f:
    config = json.load(f)

metadata = config.get('metadata', {})
print(f"✓ Loaded {len(config['nanopublications'])} access grant nanopubs to generate")
print(f"✓ Author: {metadata.get('creator_name')} ({metadata.get('creator_orcid')})")

🔨 SECTION 4: BUILD NANOPUBLICATIONS¶

def create_access_grant_nanopub(np_config, metadata):
    """
    Create an access grant nanopublication.
    
    Records that a specific DID was granted access to a dataset
    under a specific ODRL policy at a specific time.
    """
    TEMP_NP = Namespace("http://purl.org/nanopub/temp/np/")
    
    this_np = URIRef("http://purl.org/nanopub/temp/np/")
    head_graph = URIRef("http://purl.org/nanopub/temp/np/Head")
    assertion_graph = URIRef("http://purl.org/nanopub/temp/np/assertion")
    provenance_graph = URIRef("http://purl.org/nanopub/temp/np/provenance")
    pubinfo_graph = URIRef("http://purl.org/nanopub/temp/np/pubinfo")
    
    author_uri = ORCID[metadata['creator_orcid']]
    grant_uri = TEMP_NP['grant']
    target_uri = URIRef(np_config['target']['uri'])
    assignee_uri = URIRef(np_config['assignee']['did'])
    policy_uri = URIRef(np_config['under_policy'])
    
    now = datetime.now(timezone.utc).isoformat()
    
    ds = Dataset()
    
    ds.bind("this", "http://purl.org/nanopub/temp/np/")
    ds.bind("sub", TEMP_NP)
    ds.bind("np", NP)
    ds.bind("dct", DCT)
    ds.bind("nt", NT)
    ds.bind("npx", NPX)
    ds.bind("xsd", XSD)
    ds.bind("rdfs", RDFS)
    ds.bind("orcid", ORCID)
    ds.bind("prov", PROV)
    ds.bind("foaf", FOAF)
    ds.bind("odrl", ODRL)
    ds.bind("fair", FAIR)
    
    # HEAD
    head = ds.graph(head_graph)
    head.add((this_np, RDF.type, NP.Nanopublication))
    head.add((this_np, NP.hasAssertion, assertion_graph))
    head.add((this_np, NP.hasProvenance, provenance_graph))
    head.add((this_np, NP.hasPublicationInfo, pubinfo_graph))
    
    # ASSERTION — the access grant
    assertion = ds.graph(assertion_graph)
    assertion.add((grant_uri, RDF.type, ODRL.Agreement))
    assertion.add((grant_uri, ODRL.target, target_uri))
    assertion.add((grant_uri, ODRL.assignee, assignee_uri))
    assertion.add((grant_uri, PROV.generatedAtTime, Literal(now, datatype=XSD.dateTime)))
    assertion.add((grant_uri, FAIR.underPolicy, policy_uri))
    
    for action_name in np_config.get('granted_actions', ['use']):
        action_uri = ODRL_ACTIONS.get(action_name, ODRL[action_name])
        assertion.add((grant_uri, ODRL.action, action_uri))
    
    # PROVENANCE
    provenance = ds.graph(provenance_graph)
    provenance.add((assertion_graph, PROV.wasAttributedTo, author_uri))
    if np_config.get('workflow_run'):
        provenance.add((assertion_graph, PROV.wasGeneratedBy,
                       URIRef(np_config['workflow_run'])))
    
    # PUBINFO
    pubinfo = ds.graph(pubinfo_graph)
    pubinfo.add((author_uri, FOAF.name, Literal(metadata['creator_name'])))
    
    created = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%S.000Z")
    pubinfo.add((this_np, DCT.created, Literal(created, datatype=XSD.dateTime)))
    pubinfo.add((this_np, DCT.creator, author_uri))
    pubinfo.add((this_np, DCT.license, URIRef("https://creativecommons.org/licenses/by/4.0/")))
    pubinfo.add((this_np, NPX.wasCreatedAt, URIRef("https://fair2adapt-eosc.eu")))
    pubinfo.add((this_np, NPX.hasNanopubType, ODRL.Agreement))
    pubinfo.add((this_np, NPX.introduces, grant_uri))
    
    target_label = np_config['target'].get('label', np_config['target']['uri'])
    assignee_label = np_config['assignee'].get('label', np_config['assignee']['did'])
    label = f"Access grant: {target_label} → {assignee_label}"
    pubinfo.add((this_np, RDFS.label, Literal(label)))
    
    # Template references
    pubinfo.add((this_np, NT.wasCreatedFromTemplate, ODRL_GRANT_TEMPLATE))
    pubinfo.add((this_np, NT.wasCreatedFromProvenanceTemplate, PROV_TEMPLATE))
    pubinfo.add((this_np, NT.wasCreatedFromPubinfoTemplate, PUBINFO_TEMPLATE_1))
    pubinfo.add((this_np, NT.wasCreatedFromPubinfoTemplate, PUBINFO_TEMPLATE_2))
    
    return ds, label

print("✓ Function defined")

Path(OUTPUT_DIR).mkdir(parents=True, exist_ok=True)

generated_files = []

for np_config in config['nanopublications']:
    ds, label = create_access_grant_nanopub(np_config, metadata)
    
    output_file = Path(OUTPUT_DIR) / f"{np_config['id']}.trig"
    ds.serialize(destination=str(output_file), format='trig')
    generated_files.append(output_file)
    
    print(f"✓ Generated: {output_file}")

print(f"\nTotal generated: {len(generated_files)} nanopublications")

📄 SECTION 5: PREVIEW¶

if generated_files:
    print(f"Preview of {generated_files[0]}:\n")
    print("=" * 80)
    with open(generated_files[0], 'r') as f:
        print(f.read())

🚀 SECTION 6: SIGN & PUBLISH¶

PUBLISH = False
USE_TEST_SERVER = True
PROFILE_PATH = None

if PUBLISH:
    from nanopub import Nanopub, NanopubConf, load_profile
    
    profile = load_profile(PROFILE_PATH)
    print(f"Loaded profile: {profile.name}")
    
    conf = NanopubConf(profile=profile, use_test_server=USE_TEST_SERVER)
    
    for trig_file in generated_files:
        np_obj = Nanopub(rdf=trig_file, conf=conf)
        np_obj.sign()
        
        signed_path = trig_file.with_suffix('.signed.trig')
        np_obj.store(signed_path)
        print(f"✓ Signed: {signed_path}")
        
        np_obj.publish()
        print(f"✓ Published: {np_obj.source_uri}")
else:
    print("Publishing disabled. Set PUBLISH = True to enable.")