dump-things-service/dump_things_service/commands/create_merged_schema.py
Christian Monch 5bb527a8c7 chore: mark print statements as output
This removes code-linter complains by marking
print statements that produce output or error
messages as such.

It also move return-statements in the `try`-branch
of `try-except` clauses to an `else`-branch.
2026-07-01 11:13:34 +02:00

72 lines
2.3 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) # noqa T201 -- cli result output
return 0
if __name__ == '__main__':
sys.exit(main())