141 lines
4.2 KiB
Python
141 lines
4.2 KiB
Python
from __future__ import annotations
|
|
|
|
import dataclasses # noqa F401 -- used by generated code
|
|
import logging
|
|
import sys
|
|
from functools import cache
|
|
from itertools import count
|
|
from typing import (
|
|
TYPE_CHECKING,
|
|
Any,
|
|
)
|
|
from urllib.parse import urlparse
|
|
|
|
import annotated_types # noqa F401 -- used by generated code
|
|
import pydantic # noqa F401 -- used by generated code
|
|
import pydantic_core # noqa F401 -- used by generated code
|
|
from linkml.generators import (
|
|
PydanticGenerator,
|
|
PythonGenerator,
|
|
)
|
|
from linkml_runtime import SchemaView
|
|
from pydantic._internal._model_construction import ModelMetaclass
|
|
|
|
# Ensure linkml is patched
|
|
import dump_things_service.patches.enabled # noqa F401 -- apply patches
|
|
|
|
if TYPE_CHECKING:
|
|
from types import ModuleType
|
|
|
|
if (sys.version_info.major, sys.version_info.minor) == (3, 11):
|
|
sys.setrecursionlimit(2000)
|
|
|
|
lgr = logging.getLogger('dump_things_service')
|
|
|
|
serial_number = count()
|
|
_model_counter = count()
|
|
|
|
|
|
# Pydantic module generation might require a higher recursion limit than the
|
|
# default. Add a mechanism to increase it as needed, up to a maximum.
|
|
max_recursion_limit = 10000
|
|
current_recursion_limit = sys.getrecursionlimit()
|
|
|
|
|
|
def get_classes(
|
|
model: Any,
|
|
) -> list[str]:
|
|
"""get names of all subclasses of Thing"""
|
|
return get_subclasses(model, 'Thing')
|
|
|
|
|
|
def get_subclasses(
|
|
model: ModuleType,
|
|
class_name: str,
|
|
) -> list[str]:
|
|
"""get names of all subclasses (includes class_name itself)"""
|
|
|
|
# TODO: this could also be implemented via SchemaView:
|
|
# return schema_view.class_children(class_name, mixins=False)
|
|
super_class = getattr(model, class_name)
|
|
return [
|
|
name
|
|
for name, obj in model.__dict__.items()
|
|
if isinstance(obj, ModelMetaclass) and issubclass(obj, super_class)
|
|
]
|
|
|
|
|
|
# TODO: shall we use the following code?
|
|
# The code below would use schema-definitions to determine classes and not
|
|
# go through thw pydantic module generation.
|
|
@cache
|
|
def get_subclasses_2(
|
|
collection_name: str,
|
|
class_name: str,
|
|
) -> list[str]:
|
|
from dump_things_service.instance_state import get_instance_state
|
|
|
|
instance_state = get_instance_state()
|
|
schema_view = instance_state.schema_info[collection_name].schema_view
|
|
return schema_view.class_children(class_name, mixins=False)
|
|
|
|
|
|
def compile_module_with_increasing_recursion_limit(
|
|
pydantic_generator: PydanticGenerator,
|
|
schema_location: str,
|
|
) -> ModuleType:
|
|
global current_recursion_limit
|
|
|
|
module = None
|
|
module_name = (
|
|
urlparse(schema_location).path
|
|
.replace('/', '_')
|
|
.replace('-', '_')
|
|
.replace('.', '_')
|
|
)
|
|
while module is None:
|
|
try:
|
|
module = pydantic_generator.compile_module(module_name=module_name)
|
|
except RecursionError:
|
|
if current_recursion_limit >= max_recursion_limit:
|
|
lgr.error(
|
|
f'Maximum recursion limit ({max_recursion_limit}) reached when '
|
|
'building Pydantic model, cannot increase further.'
|
|
)
|
|
raise
|
|
current_recursion_limit += 1000
|
|
sys.setrecursionlimit(current_recursion_limit)
|
|
lgr.warning(
|
|
'RecursionError when building Pydantic model for schema '
|
|
f'{schema_location}, increasing recursion limit to: '
|
|
f'{current_recursion_limit}.'
|
|
)
|
|
return module
|
|
|
|
|
|
@cache
|
|
def get_model_for_schema(
|
|
schema_location: str,
|
|
) -> tuple[ModuleType, list[str], str]:
|
|
lgr.info(f'Building pydantic modulr for schema {schema_location}')
|
|
pydantic_generator = PydanticGenerator(schema_location)
|
|
model = compile_module_with_increasing_recursion_limit(
|
|
pydantic_generator,
|
|
schema_location,
|
|
)
|
|
classes = get_classes(model)
|
|
model_var_name = f'model_{next(_model_counter)}'
|
|
return model, classes, model_var_name
|
|
|
|
|
|
@cache
|
|
def get_schema_view(schema_location: str) -> SchemaView:
|
|
return SchemaView(schema_location)
|
|
|
|
|
|
@cache
|
|
def get_schema_model_for_schema(
|
|
schema_location: str,
|
|
) -> ModuleType:
|
|
lgr.info(f'Building python module for schema {schema_location}')
|
|
return PythonGenerator(schema_location).compile_module()
|