How to Check Referential Integrity

This example uses MongoDB

from linkml_store import Client

client = Client()

db = client.attach_database("mongodb://localhost:27017", "test")
db.metadata.ensure_referential_integrity = True
db.set_schema_view("../../tests/input/countries/countries.linkml.yaml")
countries_coll = db.create_collection("Country", alias="countries", recreate_if_exists=True)
routes_coll = db.create_collection("Route", alias="routes", recreate_if_exists=True)

COUNTRIES = "../../tests/input/countries/countries.jsonl"
ROUTES = "../../tests/input/countries/routes.csv"

from linkml_store.utils.format_utils import load_objects

countries = load_objects(COUNTRIES)
routes = load_objects(ROUTES)
routes

[{'origin': 'DE', 'destination': 'FR', 'method': 'rail'}]

countries_coll.insert(countries)
routes_coll.insert(routes)

routes_coll.find().rows

[{'origin': 'DE', 'destination': 'FR', 'method': 'rail'}]

for result in db.iter_validate_database():
    print(result)

Inserting invalid data

We will intentionally insert an invalid row

routes_coll.insert({"origin": "ZZZ", "destination": "YYY", "method": "rail"})

routes_coll.find().rows_dataframe

	origin	destination	method
0	DE	FR	rail
1	ZZZ	YYY	rail

results = list(db.iter_validate_database())
for result in results:
    print(result)

type='ReferentialIntegrity' severity=<Severity.ERROR: 'ERROR'> message='Referential integrity error: Country not found' instance='ZZZ' instance_index=None instantiates='Country'
type='ReferentialIntegrity' severity=<Severity.ERROR: 'ERROR'> message='Referential integrity error: Country not found' instance='YYY' instance_index=None instantiates='Country'