import copy
from datetime import datetime
from pathlib import Path
from typing import TextIO
from urllib.parse import urlparse
import yaml
from dateutil.parser import ParserError, parse
from hbreader import FileInfo, HBType, detect_type
from linkml.utils.deprecation import deprecation_warning
from linkml.utils.mergeutils import set_from_schema
from linkml_runtime.linkml_model.meta import SchemaDefinition, metamodel_version
from linkml_runtime.loaders import yaml_loader
from linkml_runtime.utils.yamlutils import YAMLMark, YAMLRoot
DATETIME_FORMAT = "%Y-%m-%dT%H:%M:%S"
yaml.error.Mark = YAMLMark
# Override the default linkml missing value tests
[docs]
def mrf(self, field_name: str) -> None:
if isinstance(self, SchemaDefinition) and field_name == "name" and self.id:
id_parts = self.id.replace("#", "/").rsplit("/")
self.name = id_parts[-1]
else:
YAMLRoot.MissingRequiredField(self, f"{type(self).__name__}.{field_name}")
SchemaDefinition.MissingRequiredField = mrf
[docs]
def load_raw_schema(
data: str | dict | TextIO | Path | SchemaDefinition,
source_file: str | None = None,
source_file_date: str | None = None,
source_file_size: int | None = None,
base_dir: str | None = None,
merge_modules: bool | None = True,
metadata: bool | None = True,
emit_metadata: bool | None = None,
) -> SchemaDefinition:
"""Load and flatten SchemaDefinition from a file name, a URL or a block of text
:param data: URL, file name, block of text, YAML object, open file handle or SchemaDefinition
:param source_file: Source file name for the schema if data is type TextIO
:param source_file_date: timestamp of source file if data is type TextIO
:param source_file_size: size of source file if data is type TextIO
:param base_dir: Working directory or base URL of sources
:param merge_modules: True means combine modules into one source, false means keep separate
:param metadata: False suppresses the source-file metadata this loader derives
(``source_file``, ``source_file_date``, ``source_file_size``, ``generation_date``).
This is a load-time suppression only — whether a generator prints such metadata is
decided by the generator. ``source_file`` is ``readonly: supplied by the schema
loader`` in the metamodel: for file/URL loads the loader-resolved path always wins
over a value embedded in the document, and ``metadata=False`` then clears it. A
caller-set ``source_file`` survives only on inputs with no source location (dict,
inline text, SchemaDefinition). See
https://github.com/linkml/linkml/issues/3699 for the ongoing work to move this
decision to serialization time.
:param emit_metadata: Legacy alias for ``metadata``; overrides it when supplied. Passing
this triggers the shared ``"metadata-flag"`` deprecation warning (see
``deprecation.py``).
:returns: Un-processed Schema Definition object
"""
def _name_from_url(url) -> str:
return urlparse(url).path.rsplit("/", 1)[-1].rsplit(".", 1)[0]
if emit_metadata is not None:
deprecation_warning("metadata-flag")
metadata = emit_metadata
# Passing a URL or file name
if detect_type(data, base_dir) not in (HBType.STRING, HBType.STRINGABLE):
assert source_file is None, "source_file parameter not allowed if data is a file or URL"
assert source_file_date is None, "source_file_date parameter not allowed if data is a file or URL"
assert source_file_size is None, "source_file_size parameter not allowed if data is a file or URL"
if isinstance(data, Path):
data = str(data)
# Records what the loader itself derived about the source. Created up front so the metadata
# handling below can rely on it regardless of which input branch was taken.
schema_metadata = FileInfo()
# Convert the input into a valid SchemaDefinition
if isinstance(data, str | dict | TextIO):
# TODO: Build a generic loader that detects type from suffix or content and invokes the appropriate loader
schema_metadata.source_file = source_file
schema_metadata.source_file_date = source_file_date
schema_metadata.source_file_size = source_file_size
schema_metadata.base_path = base_dir
schema = yaml_loader.load(
copy.deepcopy(data) if isinstance(data, dict) else data,
SchemaDefinition,
base_dir=base_dir,
metadata=schema_metadata,
)
elif isinstance(data, SchemaDefinition):
schema = copy.deepcopy(data)
else:
raise ValueError("Unrecognized input to raw loader")
if schema is None:
raise ValueError("Empty schema - cannot process")
if schema.name is None:
if schema.id is None:
raise ValueError("Unable to determine schema name")
else:
schema.name = _name_from_url(schema.id)
elif schema.id is None:
# TODO: figure out how to generate this from the default_prefix and namespace map
raise ValueError("Schema identifier must be supplied")
if metadata:
schema.source_file = schema_metadata.source_file
src_date = schema_metadata.source_file_date
try:
schema.source_file_date = parse(src_date).strftime(DATETIME_FORMAT) if src_date else None
except ParserError:
schema.source_file_date = src_date
schema.source_file_size = schema_metadata.source_file_size
schema.generation_date = datetime.now().strftime(DATETIME_FORMAT)
elif schema_metadata.source_file:
# ``metadata=False`` suppresses loader-derived source metadata. Whenever the loader had a
# source location (file, URL, or caller-supplied name), ``yaml_loader`` overwrites
# ``source_file`` with it — the slot is ``readonly: supplied by the schema loader`` — so
# the value here is always loader-derived and clearing honors the flag. When the loader
# had no source location (dict, inline text, SchemaDefinition input), nothing was
# recorded and a caller-set value survives (see in-memory import path in ``schemaloader``).
# TODO(#3699): this stripping belongs at serialization time, not load time.
schema.source_file = None
# Only set metamodel_version if the schema doesn't already define one.
# This allows schemas (like the metamodel itself) to specify their own version
# rather than inheriting from the currently installed runtime.
if not schema.metamodel_version:
schema.metamodel_version = metamodel_version
set_from_schema(schema)
return schema