reapply linting after rebase

Author: sneakers-the-rat

linkml_runtime/dumpers/delimited_file_dumper.py

@@ -5,19 +5,13 @@
 
 from json_flattener import GlobalConfig, flatten_to_csv
 from pydantic import BaseModel
-from json_flattener import GlobalConfig
 
 from linkml_runtime.dumpers.dumper_root import Dumper
 from linkml_runtime.dumpers.json_dumper import JSONDumper
 from linkml_runtime.linkml_model.meta import SchemaDefinition, SlotDefinitionName
 from linkml_runtime.utils.csvutils import get_configmap
 from linkml_runtime.utils.schemaview import SchemaView
 from linkml_runtime.utils.yamlutils import YAMLRoot
-from linkml_runtime.linkml_model.meta import SlotDefinitionName, SchemaDefinition
-from linkml_runtime.utils.schemaview import SchemaView
-
-from linkml_runtime.utils.csvutils import get_configmap
-from json_flattener import flatten_to_csv
 
 
 class DelimitedFileDumper(Dumper, ABC):

linkml_runtime/dumpers/json_dumper.py

@@ -1,5 +1,5 @@
 import json
-from datetime import datetime, date
+from datetime import date, datetime
 from decimal import Decimal
 from typing import Union
 

linkml_runtime/dumpers/rdflib_dumper.py

@@ -25,6 +25,7 @@ class RDFLibDumper(Dumper):
     This requires a SchemaView object
 
     """
+
     def as_rdf_graph(
         self,
         element: Union[BaseModel, YAMLRoot],
@@ -44,7 +45,7 @@ def as_rdf_graph(
         if isinstance(prefix_map, Converter):
             # TODO replace with `prefix_map = prefix_map.bimap` after making minimum requirement on python 3.8
             prefix_map = {record.prefix: record.uri_prefix for record in prefix_map.records}
-        logger.debug(f'PREFIXMAP={prefix_map}')
+        logger.debug(f"PREFIXMAP={prefix_map}")
         namespaces = schemaview.namespaces()
         if prefix_map:
             for k, v in prefix_map.items():
@@ -68,7 +69,9 @@ def as_rdf_graph(
         self.inject_triples(element, schemaview, g)
         return g
 
-    def inject_triples(self, element: Any, schemaview: SchemaView, graph: Graph, target_type: ElementName = None) -> Node:
+    def inject_triples(
+        self, element: Any, schemaview: SchemaView, graph: Graph, target_type: ElementName = None
+    ) -> Node:
         """
         Inject triples from conversion of element into a Graph
 
@@ -80,7 +83,7 @@ def inject_triples(self, element: Any, schemaview: SchemaView, graph: Graph, tar
         """
         namespaces = schemaview.namespaces()
         slot_name_map = schemaview.slot_name_mappings()
-        logger.debug(f'CONVERT: {element} // {type(element)} // {target_type}')
+        logger.debug(f"CONVERT: {element} // {type(element)} // {target_type}")
         if target_type in schemaview.all_enums():
             if isinstance(element, PermissibleValueText):
                 e = schemaview.get_enum(target_type)
@@ -96,22 +99,22 @@ def inject_triples(self, element: Any, schemaview: SchemaView, graph: Graph, tar
             t = schemaview.get_type(target_type)
             dt_uri = t.uri
             if dt_uri:
-                if dt_uri == 'rdfs:Resource':
+                if dt_uri == "rdfs:Resource":
                     return URIRef(schemaview.expand_curie(element))
-                elif dt_uri == 'xsd:string':
+                elif dt_uri == "xsd:string":
                     return Literal(element)
                 else:
                     if "xsd" not in namespaces:
                         namespaces["xsd"] = XSD
                     return Literal(element, datatype=namespaces.uri_for(dt_uri))
             else:
-                logger.warning(f'No datatype specified for : {t.name}, using plain Literal')
+                logger.warning(f"No datatype specified for : {t.name}, using plain Literal")
                 return Literal(element)
-        element_vars = {k: v for k, v in vars(element).items() if not k.startswith('_')}
+        element_vars = {k: v for k, v in vars(element).items() if not k.startswith("_")}
         if len(element_vars) == 0:
             id_slot = schemaview.get_identifier_slot(target_type)
             return self._as_uri(element, id_slot, schemaview)
-            #return URIRef(schemaview.expand_curie(str(element)))
+            # return URIRef(schemaview.expand_curie(str(element)))
         element_type = type(element)
         cn = element_type.class_name
         id_slot = schemaview.get_identifier_slot(cn)
@@ -134,7 +137,7 @@ def inject_triples(self, element: Any, schemaview: SchemaView, graph: Graph, tar
                 if k in slot_name_map:
                     k = slot_name_map[k].name
                 else:
-                    logger.error(f'Slot {k} not in name map')
+                    logger.error(f"Slot {k} not in name map")
                 slot = schemaview.induced_slot(k, cn)
                 if not slot.identifier:
                     slot_uri = URIRef(schemaview.get_uri(slot, expand=True))
@@ -151,7 +154,7 @@ def dump(
         element: Union[BaseModel, YAMLRoot],
         to_file: str,
         schemaview: SchemaView = None,
-        fmt: str = 'turtle',
+        fmt: str = "turtle",
         prefix_map: Union[dict[str, str], Converter, None] = None,
         **args,
     ) -> None:
@@ -171,7 +174,7 @@ def dumps(
         self,
         element: Union[BaseModel, YAMLRoot],
         schemaview: SchemaView = None,
-        fmt: Optional[str] = 'turtle',
+        fmt: Optional[str] = "turtle",
         prefix_map: Union[dict[str, str], Converter, None] = None,
     ) -> str:
         """
@@ -183,12 +186,10 @@ def dumps(
         :param prefix_map:
         :return: serialization of rdflib Graph containing element
         """
-        return self.as_rdf_graph(element, schemaview, prefix_map=prefix_map).\
-            serialize(format=fmt)
+        return self.as_rdf_graph(element, schemaview, prefix_map=prefix_map).serialize(format=fmt)
 
     def _as_uri(self, element_id: str, id_slot: Optional[SlotDefinition], schemaview: SchemaView) -> URIRef:
         if id_slot and schemaview.is_slot_percent_encoded(id_slot):
             return URIRef(urllib.parse.quote(element_id))
         else:
             return schemaview.namespaces().uri_for(element_id)
-

linkml_runtime/linkml_model/linkml_files.py

@@ -24,7 +24,8 @@ def _generate_next_value_(name, start, count, last_values):
 
 
 class Source(_AutoName):
-    """ LinkML package source name """
+    """LinkML package source name"""
+
     META = auto()
     TYPES = auto()
     MAPPINGS = auto()
@@ -33,7 +34,8 @@ class Source(_AutoName):
 
 
 class Format(_AutoName):
-    """ LinkML package formats """
+    """LinkML package formats"""
+
     EXCEL = auto()
     GRAPHQL = auto()
     JSON = auto()
@@ -55,48 +57,51 @@ class Format(_AutoName):
     SQLSCHEMA = auto()
     YAML = auto()
 
+
 @dataclass
 class FormatPath:
     path: str
     extension: str
 
-    def model_path(self, model:str) -> Path:
+    def model_path(self, model: str) -> Path:
         return (Path(self.path) / model).with_suffix(self.extension)
 
+
 class _Path:
-    """ LinkML Relative paths"""
-    EXCEL = FormatPath("excel","xlsx" )
-    GRAPHQL = FormatPath("graphql","graphql" )
-    JSON = FormatPath("json","json" )
-    JSONLD = FormatPath("jsonld","context.jsonld" )
-    JSON_SCHEMA = FormatPath("jsonschema", "schema.json" )
-    NATIVE_JSONLD = FormatPath("jsonld", "context.jsonld" )
-    NATIVE_RDF = FormatPath("rdf","ttl" )
-    NATIVE_SHEXC = FormatPath("shex","shex" )
-    NATIVE_SHEXJ = FormatPath("shex","shexj" )
-    OWL = FormatPath("owl","owl.ttl" )
-    PREFIXMAP = FormatPath('prefixmap','yaml' )
-    PROTOBUF = FormatPath("protobuf","proto" )
-    PYTHON = FormatPath("","py" )
-    RDF = FormatPath("rdf","ttl" )
-    SHACL = FormatPath("shacl","shacl.ttl" )
-    SHEXC = FormatPath("shex","shex" )
-    SHEXJ = FormatPath("shex","shexj" )
-    SQLDDL = FormatPath("sqlddl","sql" )
-    SQLSCHEMA = FormatPath("sqlschema","sql" )
-    YAML = FormatPath((Path("model") / "schema").as_posix(),"yaml" )
+    """LinkML Relative paths"""
+
+    EXCEL = FormatPath("excel", "xlsx")
+    GRAPHQL = FormatPath("graphql", "graphql")
+    JSON = FormatPath("json", "json")
+    JSONLD = FormatPath("jsonld", "context.jsonld")
+    JSON_SCHEMA = FormatPath("jsonschema", "schema.json")
+    NATIVE_JSONLD = FormatPath("jsonld", "context.jsonld")
+    NATIVE_RDF = FormatPath("rdf", "ttl")
+    NATIVE_SHEXC = FormatPath("shex", "shex")
+    NATIVE_SHEXJ = FormatPath("shex", "shexj")
+    OWL = FormatPath("owl", "owl.ttl")
+    PREFIXMAP = FormatPath("prefixmap", "yaml")
+    PROTOBUF = FormatPath("protobuf", "proto")
+    PYTHON = FormatPath("", "py")
+    RDF = FormatPath("rdf", "ttl")
+    SHACL = FormatPath("shacl", "shacl.ttl")
+    SHEXC = FormatPath("shex", "shex")
+    SHEXJ = FormatPath("shex", "shexj")
+    SQLDDL = FormatPath("sqlddl", "sql")
+    SQLSCHEMA = FormatPath("sqlschema", "sql")
+    YAML = FormatPath((Path("model") / "schema").as_posix(), "yaml")
 
     @classmethod
     def items(cls) -> dict[str, FormatPath]:
-        return {k:v for k,v in cls.__dict__.items() if not k.startswith('_')}
+        return {k: v for k, v in cls.__dict__.items() if not k.startswith("_")}
 
     @classmethod
-    def get(cls, item:Union[str,Format]) -> FormatPath:
+    def get(cls, item: Union[str, Format]) -> FormatPath:
         if isinstance(item, Format):
             item = item.name.upper()
         return getattr(cls, item)
 
-    def __class_getitem__(cls, item:str) -> FormatPath:
+    def __class_getitem__(cls, item: str) -> FormatPath:
         return getattr(cls, item)
 
 
@@ -108,14 +113,15 @@ def __class_getitem__(cls, item:str) -> FormatPath:
     Format.PROTOBUF,
     Format.SHACL,
     Format.SQLDDL,
-    Format.SQLSCHEMA
+    Format.SQLSCHEMA,
 )
 
 
 class ReleaseTag(_AutoName):
-    """ Release tags
+    """Release tags
     LATEST - the absolute latest in the supplied branch
-    CURRENT - the latest _released_ version in the supplied branch """
+    CURRENT - the latest _released_ version in the supplied branch"""
+
     LATEST = auto()
     CURRENT = auto()
 
@@ -137,12 +143,12 @@ def _build_path(source: Source, fmt: Format) -> PathParts:
 def _build_loc(base: str, source: Source, fmt: Format) -> str:
     """A github location"""
     # urls are always forward slash separated, so hardcoding is appropriate here
-    path = '/'.join(_build_path(source, fmt))
-    return urljoin(base, path).replace('blob/', '')
+    path = "/".join(_build_path(source, fmt))
+    return urljoin(base, path).replace("blob/", "")
 
 
 def URL_FOR(source: Source, fmt: Format) -> str:
-    """ Return the URL to retrieve source in format """
+    """Return the URL to retrieve source in format"""
     fmt_path: FormatPath = _Path.get(fmt.name)
     return f"{LINKML_URL_BASE}{source.value}.{fmt_path.extension}"
 
@@ -152,14 +158,16 @@ def LOCAL_PATH_FOR(source: Source, fmt: Format) -> str:
 
 
 def GITHUB_IO_PATH_FOR(source: Source, fmt: Format, version="latest") -> str:
-    path = '/'.join([version, 'linkml_model', *_build_path(source, fmt)])
+    path = "/".join([version, "linkml_model", *_build_path(source, fmt)])
     return urljoin(GITHUB_IO_BASE, path)
 
 
-def GITHUB_PATH_FOR(source: Source,
-                    fmt: Format,
-                    release: Optional[Union[ReleaseTag, str]] = ReleaseTag.CURRENT,
-                    branch: Optional[str] = "main") -> str:
+def GITHUB_PATH_FOR(
+    source: Source,
+    fmt: Format,
+    release: Optional[Union[ReleaseTag, str]] = ReleaseTag.CURRENT,
+    branch: Optional[str] = "main",
+) -> str:
     def do_request(url) -> object:
         resp = requests.get(url)
         if resp.ok:
@@ -169,7 +177,7 @@ def do_request(url) -> object:
     def tag_to_commit(tag: str) -> str:
         tags = do_request(f"{GITHUB_API_BASE}tags?per_page=100")
         for tagent in tags:
-            if tagent['name'] == tag:
+            if tagent["name"] == tag:
                 return _build_loc(f"{GITHUB_BASE}blob/{tagent['commit']['sha']}/", source, fmt)
         raise ValueError(f"Tag: {tag} not found!")
 
@@ -178,13 +186,13 @@ def tag_to_commit(tag: str) -> str:
 
     # Return the absolute latest entry for branch
     if release is ReleaseTag.LATEST or (release is ReleaseTag.CURRENT and branch != "main"):
-        path = '/'.join([branch, 'linkml_model', *_build_path(source, fmt)])
+        path = "/".join([branch, "linkml_model", *_build_path(source, fmt)])
         return urljoin(GITHUB_BASE, path)
 
     # Return the latest published version
     elif release is ReleaseTag.CURRENT:
         release = do_request(f"{GITHUB_API_BASE}releases/latest")
-        return tag_to_commit(release['tag_name'])
+        return tag_to_commit(release["tag_name"])
 
     # Return a specific tag
     else:
@@ -212,7 +220,9 @@ def url(self) -> str:
         def file(self) -> str:
             return LOCAL_PATH_FOR(self._model, self._format)
 
-        def github_loc(self, tag: Optional[str] = None, branch: Optional[str] = None, release: ReleaseTag = None) -> str:
+        def github_loc(
+            self, tag: Optional[str] = None, branch: Optional[str] = None, release: ReleaseTag = None
+        ) -> str:
             if not tag and not branch and not release:
                 return GITHUB_IO_PATH_FOR(self._model, self._format)
             if tag:

linkml_runtime/loaders/json_loader.py

@@ -1,39 +1,45 @@
 import json
 import logging
 from pathlib import Path
-from typing import Union, TextIO, Optional
+from typing import Optional, TextIO, Union
 
 from hbreader import FileInfo
+from pydantic import BaseModel
 
 from linkml_runtime.loaders.loader_root import Loader
 from linkml_runtime.utils.yamlutils import YAMLRoot
-from pydantic import BaseModel
 
 logger = logging.getLogger(__name__)
 
 
 class JSONLoader(Loader):
 
-    def load_as_dict(self, 
-                     source: Union[str, dict, TextIO],
-                     *, 
-                     base_dir: Optional[str] = None,
-                     metadata: Optional[FileInfo] = None) -> Union[dict, list[dict]]:
-        data = self._read_source(source, base_dir=base_dir, metadata=metadata, accept_header="application/ld+json, application/json, text/json")
+    def load_as_dict(
+        self, source: Union[str, dict, TextIO], *, base_dir: Optional[str] = None, metadata: Optional[FileInfo] = None
+    ) -> Union[dict, list[dict]]:
+        data = self._read_source(
+            source,
+            base_dir=base_dir,
+            metadata=metadata,
+            accept_header="application/ld+json, application/json, text/json",
+        )
         data_as_dict = json.loads(data) if isinstance(data, str) else data
         return self.json_clean(data_as_dict)
 
-    def load_any(self, 
-                 source: Union[str, dict, TextIO, Path],
-                 target_class: type[Union[BaseModel, YAMLRoot]],
-                 *, 
-                 base_dir: Optional[str] = None,
-                 metadata: Optional[FileInfo] = None, 
-                 **_) -> Union[BaseModel, YAMLRoot, list[BaseModel], list[YAMLRoot]]:
+    def load_any(
+        self,
+        source: Union[str, dict, TextIO, Path],
+        target_class: type[Union[BaseModel, YAMLRoot]],
+        *,
+        base_dir: Optional[str] = None,
+        metadata: Optional[FileInfo] = None,
+        **_,
+    ) -> Union[BaseModel, YAMLRoot, list[BaseModel], list[YAMLRoot]]:
         """
         Load the JSON in source into the python target_class structure
 
-        :param source: JSON data source. Can be a URL, a file name, a JSON string, a resolveable filepath or an existing graph
+        :param source: JSON data source. Can be a URL, a file name, a JSON string, a resolveable filepath,
+            or an existing graph
         :param target_class: LinkML class to load the JSON into
         :param base_dir: Base directory that can be used if file name or URL.  This is copied into metadata if present
         :param metadata: source information. Used by some loaders to record where information came from
@@ -54,7 +60,7 @@ def load_any(self,
         data_as_dict = self.load_as_dict(source, base_dir=base_dir, metadata=metadata)
 
         if isinstance(data_as_dict, dict):
-            typ = data_as_dict.pop('@type', None)
+            typ = data_as_dict.pop("@type", None)
             if typ and typ != target_class.__name__:
                 logger.warning(f"Warning: input type mismatch. Expected: {target_class.__name__}, Actual: {typ}")