feat(generate): ✨Implement code generation for Python (#27)

Part of #1.
The part with the `sqlmodel`-parser was done in co-development with
@DeltaDaniel.

---------

Co-authored-by: kevin <[email protected]>

.pre-commit-config.yaml

@@ -7,12 +7,12 @@ repos:
       - id: end-of-file-fixer
       - id: trailing-whitespace
   - repo: https://github.com/psf/black
-    rev: 23.9.1
+    rev: 24.8.0
     hooks:
       - id: black
         language_version: python3
   - repo: https://github.com/pycqa/isort
-    rev: 5.12.0
+    rev: 5.13.2
     hooks:
       - id: isort
         name: isort (python)

pyproject.toml

@@ -23,6 +23,8 @@ dependencies = [
     "httpx",
     "datamodel-code-generator",
     "autoflake",
+    "more_itertools",
+    "sqlmodel",
 ]     # add all the dependencies here
 dynamic = ["readme", "version"]
 

requirements.txt

@@ -43,6 +43,8 @@ email-validator==2.2.0
     # via pydantic
 genson==1.3.0
     # via datamodel-code-generator
+greenlet==3.1.1
+    # via sqlalchemy
 h11==0.14.0
     # via httpcore
 httpcore==1.0.5
@@ -67,6 +69,8 @@ markupsafe==2.1.5
     # via jinja2
 mdurl==0.1.2
     # via markdown-it-py
+more-itertools==10.5.0
+    # via bo4e-cli (pyproject.toml)
 mypy-extensions==1.0.0
     # via black
 packaging==24.1
@@ -83,6 +87,7 @@ pydantic[email]==2.9.0
     # via
     #   bo4e-cli (pyproject.toml)
     #   datamodel-code-generator
+    #   sqlmodel
 pydantic-core==2.23.2
     # via pydantic
 pyflakes==3.2.0
@@ -107,13 +112,18 @@ sniffio==1.3.1
     # via
     #   anyio
     #   httpx
+sqlalchemy==2.0.35
+    # via sqlmodel
+sqlmodel==0.0.22
+    # via bo4e-cli (pyproject.toml)
 typer==0.12.5
     # via bo4e-cli (pyproject.toml)
 typing-extensions==4.12.2
     # via
     #   pydantic
     #   pydantic-core
     #   pygithub
+    #   sqlalchemy
     #   typer
 tzdata==2024.1
     # via pydantic

src/bo4e_cli/commands/generate.py

@@ -2,24 +2,16 @@
 This module contains the generate command.
 """
 
-from enum import StrEnum
 from pathlib import Path
 from typing import Annotated
 
 import typer
 
-from bo4e_cli.commands.dummy import dummy
 from bo4e_cli.commands.entry import app
-
-
-class GenerateType(StrEnum):
-    """
-    A custom type for the generate command.
-    """
-
-    PYTHON_PYDANTIC_V1 = "python-pydantic-v1"
-    PYTHON_PYDANTIC_V2 = "python-pydantic-v2"
-    PYTHON_SQL_MODEL = "python-sql-model"
+from bo4e_cli.generate.python.entry import generate_bo4e_schemas
+from bo4e_cli.io.console import CONSOLE
+from bo4e_cli.io.schemas import read_schemas
+from bo4e_cli.types import GenerateType
 
 
 @app.command()
@@ -44,4 +36,9 @@ def generate(
 
     Several output types are available, see --output-type.
     """
-    dummy(input_dir=input_dir, output_dir=output_dir, output_type=output_type, clear_output=clear_output)
+    schemas = read_schemas(input_dir)
+    if output_type in (GenerateType.PYTHON_PYDANTIC_V1, GenerateType.PYTHON_PYDANTIC_V2, GenerateType.PYTHON_SQL_MODEL):
+        generate_bo4e_schemas(schemas, output_dir, output_type, clear_output)
+    else:
+        CONSOLE.print(f"Output type {output_type} is not supported yet.")
+        raise typer.Exit(1)

src/bo4e_cli/edit/update_refs.py

@@ -9,7 +9,8 @@
 from bo4e_cli.io.console import CONSOLE
 from bo4e_cli.io.github import OWNER, REPO
 from bo4e_cli.models.meta import SchemaMeta, Schemas
-from bo4e_cli.models.schema import AllOf, AnyOf, Array, Object, Reference, SchemaType
+from bo4e_cli.models.schema import Reference
+from bo4e_cli.utils.fields import iter_schema_type
 
 REF_ONLINE_REGEX = re.compile(
     rf"^https://raw\.githubusercontent\.com/(?:{OWNER.upper()}|{OWNER.lower()}|{OWNER.capitalize()}|Hochfrequenz)/"
@@ -33,7 +34,7 @@ def update_reference(
     Example of reference to definitions:
     #/$defs/Angebot
     """
-    schema_cls_namespace = schemas.search_index_by_cls_name
+    schema_cls_namespace = schemas.names
     match = REF_ONLINE_REGEX.search(field.ref)
     if match is not None:
         CONSOLE.print(f"Matched online reference: {field.ref}", show_only_on_verbose=True)
@@ -80,34 +81,8 @@ def update_references(schema: SchemaMeta, schemas: Schemas) -> None:
     on every Reference object.
     """
 
-    def update_or_iter(_object: SchemaType) -> None:
-        if isinstance(_object, Object):
-            iter_object(_object)
-        elif isinstance(_object, AnyOf):
-            iter_any_of(_object)
-        elif isinstance(_object, AllOf):
-            iter_all_of(_object)
-        elif isinstance(_object, Array):
-            iter_array(_object)
-        elif isinstance(_object, Reference):
-            update_reference(_object, schema, schemas)
-
-    def iter_object(_object: Object) -> None:
-        for prop in _object.properties.values():
-            update_or_iter(prop)
-
-    def iter_any_of(_object: AnyOf) -> None:
-        for item in _object.any_of:
-            update_or_iter(item)
-
-    def iter_all_of(_object: AllOf) -> None:
-        for item in _object.all_of:
-            update_or_iter(item)
-
-    def iter_array(_object: Array) -> None:
-        update_or_iter(_object.items)
-
-    update_or_iter(schema.schema_parsed)
+    for reference in iter_schema_type(schema.schema_parsed, Reference):
+        update_reference(reference, schema, schemas)
 
 
 def update_references_all_schemas(schemas: Schemas) -> None:

src/bo4e_cli/generate/python/custom_templates/BaseModel.jinja2

@@ -0,0 +1,61 @@
+{%- if SQL %}
+
+{%- for import in SQL.imports %}
+{%- if import.alias is not none %}
+from {{ import.from_ }} import {{ import.import_ }} as {{ import.alias }}
+{%- else %}
+from {{ import.from_ }} import {{ import.import_ }}
+{%- endif %}
+{%- endfor -%}
+
+{%- endif %}
+
+{% for decorator in decorators -%}
+{{ decorator }}
+{% endfor -%}
+class {{ class_name }}({% if base_class %}{{ base_class }}{% endif %}{% if SQL %}, table=True{% endif %}):{% if comment is defined %}  # {{ comment }}{% endif %}
+{%- if description %}
+    """
+    {{ description | indent(4) }}
+    """
+{%- endif %}
+{%- if not fields and not description %}
+    pass
+{%- endif %}
+{%- if config %}
+{%- filter indent(4) %}
+{% include 'Config.jinja2' %}
+{%- endfilter %}
+{%- endif %}
+{%- for field in fields -%}
+    {%- if not field.annotated and field.field %}
+    {{ field.name }}: {{ field.type_hint }} = {{ field.field }}
+    {%- else %}
+    {%- if field.annotated %}
+    {{ field.name }}: {{ field.annotated }}
+    {%- else %}
+    {{ field.name }}: {{ field.type_hint }}
+    {%- endif %}
+    {%- if not (field.required or (field.represented_default == 'None' and field.strip_default_none))
+            %} = {{ field.represented_default }}
+    {%- endif -%}
+    {%- endif %}
+    {%- if field.docstring %}
+    """
+    {{ field.docstring | indent(4) }}
+    """
+    {%- endif %}
+{%- for method in methods -%}
+    {{ method }}
+{%- endfor -%}
+{%- endfor -%}
+{%- if SQL and SQL.fields %}
+{%- for field_name, field in SQL.fields.items() %}
+    {{ field_name }}: {{ field.annotation }} = {{ field.definition }}
+    {%- if field.description %}
+    """
+    {{ field.description | indent(4) }}
+    """
+    {%- endif %}
+{%- endfor -%}
+{%- endif %}

src/bo4e_cli/generate/python/custom_templates/Config.jinja2

@@ -0,0 +1,6 @@
+model_config = SQLModelConfig(
+    arbitrary_types_allowed=True,
+{%- for field_name, value in config.dict(exclude_unset=True).items() %}
+    {{ field_name }}={{ value }},
+{%- endfor %}
+)

src/bo4e_cli/generate/python/custom_templates/Enum.jinja2

@@ -0,0 +1,17 @@
+{% for decorator in decorators -%}
+{{ decorator }}
+{% endfor -%}
+class {{ class_name }}({{ base_class }}):
+{%- if description %}
+    """
+    {{ description | indent(4) }}
+    """
+{%- endif %}
+{%- for field in fields %}
+    {{ field.name[:63] }} = {{ field.default }}
+    {%- if field.docstring %}
+    """
+    {{ field.docstring | indent(4) }}
+    """
+    {%- endif %}
+{%- endfor -%}

src/bo4e_cli/generate/python/custom_templates/ManyLinks.jinja2

@@ -0,0 +1,19 @@
+"""
+File containing all linking classes for many-many relations in the BO4E version
+"""
+import uuid as uuid_pkg
+
+from sqlalchemy import Column, ForeignKey
+from sqlmodel import Field, SQLModel
+
+{%- for link in links %}
+class {{ link.table_name }}(SQLModel, table=True):
+    """
+    class linking m-n relation of tables {{ link.cls1 }} and {{ link.cls2 }} for field {{ link.rel_field_name1 }}.
+    """
+    {{ link.id_field_name1 }}: uuid_pkg.UUID = Field(..., primary_key=True, foreign_key="{{ link.cls1.lower() }}.id", ondelete="CASCADE")
+    """Id linking to {{ link.cls1 }}."""
+    {{ link.id_field_name2 }}: uuid_pkg.UUID = Field(..., primary_key=True, foreign_key="{{ link.cls2.lower() }}.id", ondelete="CASCADE")
+    """Id linking to {{ link.cls2 }}."""
+
+{%- endfor -%}

src/bo4e_cli/generate/python/entry.py

@@ -2,61 +2,35 @@
 This module is the entry point for the CLI bo4e-generator.
 """
 
-import shutil
 from pathlib import Path
-from typing import Optional
+from typing import Literal
 
-from bo4e_cli.generate.python.parser import (
-    OutputType,
-    bo4e_init_file_content,
-    bo4e_version_file_content,
-    get_formatter,
-    parse_bo4e_schemas,
-)
-from bo4e_cli.generate.python.schema import get_namespace, get_version
-from bo4e_cli.generate.python.sqlparser import remove_unused_imports
-
-
-def resolve_paths(input_directory: Path, output_directory: Path) -> tuple[Path, Path]:
-    """
-    Resolve the input and output paths. The data-model-parser have problems with handling relative paths.
-    """
-    if not input_directory.is_absolute():
-        input_directory = input_directory.resolve()
-    if not output_directory.is_absolute():
-        output_directory = output_directory.resolve()
-    return input_directory, output_directory
+from bo4e_cli.generate.python.format import post_process_files
+from bo4e_cli.generate.python.parser import bo4e_init_file_content, bo4e_version_file_content, parse_bo4e_schemas
+from bo4e_cli.io.cleanse import clear_dir_if_needed
+from bo4e_cli.io.file import write_file_contents
+from bo4e_cli.models.meta import Schemas
+from bo4e_cli.types import GenerateType
 
 
 def generate_bo4e_schemas(
-    input_directory: Path,
+    schemas: Schemas,
     output_directory: Path,
-    output_type: OutputType,
-    clear_output: bool = False,
-    target_version: Optional[str] = None,
+    generate_type: Literal[
+        GenerateType.PYTHON_PYDANTIC_V1,
+        GenerateType.PYTHON_PYDANTIC_V2,
+        GenerateType.PYTHON_SQL_MODEL,
+    ],
+    clear_output: bool = True,
 ) -> None:
     """
     Generate all BO4E schemas from the given input directory and save them in the given output directory.
     """
-    input_directory, output_directory = resolve_paths(input_directory, output_directory)
-    namespace = get_namespace(input_directory)
-    file_contents = parse_bo4e_schemas(input_directory, namespace, output_type)
-    version = get_version(target_version, namespace)
-    file_contents[Path("__version__.py")] = bo4e_version_file_content(version)
-    file_contents[Path("__init__.py")] = bo4e_init_file_content(namespace, version)
-    if clear_output and output_directory.exists():
-        shutil.rmtree(output_directory)
+    file_contents = parse_bo4e_schemas(schemas, generate_type)
+    file_contents[Path("__version__.py")] = bo4e_version_file_content(schemas.version.to_str_without_prefix())
+    file_contents[Path("__init__.py")] = bo4e_init_file_content(schemas)
 
-    formatter = get_formatter()
-    for relative_file_path, file_content in file_contents.items():
-        file_path = output_directory / relative_file_path
-        file_path.parent.mkdir(parents=True, exist_ok=True)
-        if (
-            relative_file_path.name not in ["__init__.py", "__version__.py"]
-            and OutputType[output_type] == OutputType.SQL_MODEL
-        ):
-            file_content = remove_unused_imports(file_content)
-        file_content = formatter.format_code(file_content)
-        file_path.write_text(file_content, encoding="utf-8")
-        print(f"Created {file_path}")
-    print("Done.")
+    post_process_files(file_contents)
+    if clear_output:
+        clear_dir_if_needed(output_directory)
+    write_file_contents(file_contents, base_path=output_directory)