format

src/linkml_transformer/compiler/sql_compiler.py

@@ -2,9 +2,11 @@
 from linkml_runtime.linkml_model import SlotDefinition
 
 from linkml_transformer.compiler.compiler import CompiledSpecification, Compiler
-from linkml_transformer.datamodel.transformer_model import TransformationSpecification, ClassDerivation, \
-    SerializationSyntaxType
-
+from linkml_transformer.datamodel.transformer_model import (
+    ClassDerivation,
+    SerializationSyntaxType,
+    TransformationSpecification,
+)
 
 LINKML_TO_SQL_TYPE_MAP = {
     "string": "TEXT",
@@ -16,17 +18,17 @@
     "date": "DATE",
     "time": "TIME",
     "uri": "TEXT",
-    "any": "TEXT"
+    "any": "TEXT",
 }
 
 
-
 class SQLCompiler(Compiler):
     """
     Compiles a Transformation Specification to SQL CREATE TABLE or VIEW statements.
 
     Note: this is currently highly geared towards DuckDB.
     """
+
     add_if_not_exists: bool = True
     new_table_when_transforming: bool = False
 
@@ -36,7 +38,12 @@ def compile(self, specification: TransformationSpecification) -> CompiledSpecifi
             self.compile_class(compiled, cd, specification)
         return compiled
 
-    def compile_class(self, compiled: CompiledSpecification, cd: ClassDerivation, specification: TransformationSpecification) -> None:
+    def compile_class(
+        self,
+        compiled: CompiledSpecification,
+        cd: ClassDerivation,
+        specification: TransformationSpecification,
+    ) -> None:
         """
         Compile a class derivation to SQL.
 
@@ -81,7 +88,6 @@ def compile_slot_derivation(self, sd) -> str:
                     expr = f"STRING_AGG({expr}, '{delimiter}')"
         return f"  {sd.name} AS {expr}"
 
-
     def create_ddl(self, schemaview: SchemaView) -> str:
         """
         Create DDL for the entire schema.
@@ -130,5 +136,3 @@ def sql_type(self, slot: SlotDefinition, schemaview: SchemaView) -> str:
         if slot.multivalued:
             typ = f"{typ}[]"
         return typ
-
-

src/linkml_transformer/session.py

@@ -51,7 +51,6 @@ def set_transformer_specification(
                 obj = yaml.safe_load(open(specification))
             self.set_transformer_specification(obj)
 
-
     def set_source_schema(self, schema: Union[str, Path, dict, SchemaView, SchemaDefinition]):
         """
         Sets the schema from a path or SchemaView object.
@@ -72,7 +71,10 @@ def set_source_schema(self, schema: Union[str, Path, dict, SchemaView, SchemaDef
         self._target_schema = None
 
     def set_object_transformer(
-        self, transformer: Optional[Union[ObjectTransformer, TransformationSpecification, dict, str, Path]] = None
+        self,
+        transformer: Optional[
+            Union[ObjectTransformer, TransformationSpecification, dict, str, Path]
+        ] = None,
     ):
         if transformer is None:
             if self.object_transformer is not None:
@@ -95,9 +97,7 @@ def target_schema(self) -> SchemaDefinition:
         if self._target_schema is None:
             if not self.schema_mapper:
                 self.schema_mapper = SchemaMapper(source_schemaview=self.source_schemaview)
-            self._target_schema = self.schema_mapper.derive_schema(
-                self.transformer_specification
-            )
+            self._target_schema = self.schema_mapper.derive_schema(self.transformer_specification)
         return self._target_schema
 
     @property

src/linkml_transformer/transformer/duckdb_transformer.py

@@ -3,7 +3,6 @@
 from dataclasses import dataclass
 from typing import Any, Dict, List, Optional, Type, Union
 
-
 from linkml_transformer.transformer.transformer import OBJECT_TYPE, Transformer
 
 DICT_OBJ = Dict[str, Any]
@@ -33,5 +32,6 @@ def transform(
         :return: transformed data, either as type target_type or a dictionary
         """
         import duckdb
+
         sv = self.source_schemaview
         raise NotImplementedError("DuckDBTransformer.transform")

tests/test_compiler/test_duckdb_compiler.py

@@ -35,8 +35,8 @@ def test_compile(session):
     target_ddl = compiler.create_ddl(target_sv)
     print(target_ddl)
 
-
     import duckdb
-    conn = duckdb.connect(':memory:')
+
+    conn = duckdb.connect(":memory:")
     conn.execute(target_ddl)
     conn.execute(compiled.serialization)