dump-things-service/dump_things_service/commands/create_merged_schema.py
Michael Hanke c332abbd1b
Some checks failed
Codespell / Check for spelling errors (push) Successful in 42s
Codespell / Check for spelling errors (pull_request) Successful in 43s
Ruff / Code linting (push) Failing after 52s
Ruff / Code linting (pull_request) Failing after 54s
Test execution / Test-all (push) Successful in 1m27s
Type annotation (PR) / check (pull_request) Failing after 1m30s
chore: auto-apply code formatting fixes via hatch check fmt --fix
2026-06-30 20:41:05 +02:00

72 lines
2.2 KiB
Python

import sys
from argparse import ArgumentParser
import yaml
from linkml_runtime.utils.schemaview import SchemaView
# Patch linkml
from dump_things_service.patches import enabled # noqa: F401 -- patches LinkML
parser = ArgumentParser(
prog='Create a static schema with all imported schemas integrated',
)
parser.add_argument('schema', help='File containing a schema definition')
def update_uris_for_elements(
all_elements: dict, attribute_name: str, prefix_index: dict
):
for name, info in all_elements.items():
uri = getattr(info, attribute_name)
if uri is None:
source = info.from_schema + '/'
if source in prefix_index:
source = prefix_index[source] + ':'
uri = source + name
setattr(info, attribute_name, uri)
def update_uris(schema_view: SchemaView):
"""Update element-defining URIs to the original element source
Element-defining URIs (e.g., slot_uri, class_uri) are by default set to
the schema in which the element is defined. In this case, that would be the
root-schema of the merged schema. The merged schema would then contain different
element-defining URIs than the individual schemas that were merged. That
means that records that are based on the individual schemas would not be
valid in the merged schema.
This function updates the element-defining URIs to the original source.
"""
prefix_index = {
p.prefix_reference: p.prefix_prefix
for p in schema_view.schema.prefixes.values()
}
update_uris_for_elements(schema_view.schema.slots, 'slot_uri', prefix_index)
update_uris_for_elements(schema_view.schema.types, 'uri', prefix_index)
update_uris_for_elements(schema_view.schema.enums, 'enum_uri', prefix_index)
update_uris_for_elements(schema_view.schema.classes, 'class_uri', prefix_index)
def main():
args = parser.parse_args()
schema_view = SchemaView(
schema=args.schema,
importmap=None,
merge_imports=True,
)
update_uris(schema_view)
text = yaml.dump(
schema_view.schema,
Dumper=yaml.SafeDumper,
allow_unicode=True,
sort_keys=False,
)
print(text)
return 0
if __name__ == '__main__':
sys.exit(main())