Added cpl generate
All checks were successful
Test before pr merge / test-lint (pull_request) Successful in 7s

This commit is contained in:
2025-10-16 12:46:11 +02:00
parent 8d0bc13cc0
commit 98ed458d7c
20 changed files with 362 additions and 1 deletions

View File

@@ -15,7 +15,7 @@
},
"references": [],
"main": "cpl/cli/main.py",
"directory": "cpl",
"directory": "cpl/cli",
"build": {
"include": [
"_templates/"

View File

@@ -0,0 +1,9 @@
from abc import ABC
class <Name>ABC(ABC):
def __init__(self):
ABC.__init__(self)
print("<schematic> <multi_camelName> initialized")

View File

@@ -0,0 +1,16 @@
from cpl.application.abc import ApplicationABC
from cpl.core.environment import Environment
from cpl.core.log import LoggerABC
from cpl.dependency import ServiceProvider
from cpl.dependency.typing import Modules
class <Name>(ApplicationABC):
def __init__(self, services: ServiceProvider, modules: Modules):
ApplicationABC.__init__(self, services, modules)
self._logger = services.get_service(LoggerABC)
async def main(self):
self._logger.debug(f"Host: {Environment.get_host_name()}")
self._logger.debug(f"Environment: {Environment.get_environment()}")

View File

@@ -0,0 +1,10 @@
from cpl.core.configuration import ConfigurationModelABC
class <Name>Config(ConfigurationModelABC):
def __init__(
self,
src: dict = None,
):
ConfigurationModelABC.__init__(self, src)

View File

@@ -0,0 +1,9 @@
from cpl.database.abc import DbModelDaoABC
class <Name>Dao(DbModelDaoABC[<Name>]):
def __init__(self):
DbModelDaoABC.__init__(self, <Name>, "<multi_name>")
self.attribute(<Name>.name, str)

View File

@@ -0,0 +1,23 @@
from datetime import datetime
from typing import Self
from cpl.core.typing import SerialId
from cpl.database.abc import DbModelABC
class <Name>(DbModelABC[Self]):
def __init__(
self,
id: SerialId,
name: str,
deleted: bool = False,
editor_id: SerialId | None = None,
created: datetime | None = None,
updated: datetime | None = None,
):
DbModelABC.__init__(self, id, deleted, editor_id, created, updated)
self._name = name
@property
def name(self) -> str:
return self._name

View File

@@ -0,0 +1,29 @@
from datetime import datetime
from typing import Self
from cpl.core.typing import SerialId
from cpl.database.abc import DbJoinModelABC
class <Name>Join(DbJoinModelABC[Self]):
def __init__(
self,
id: SerialId,
source_id: SerialId,
reference_id: SerialId,
deleted: bool = False,
editor_id: SerialId | None = None,
created: datetime | None = None,
updated: datetime | None = None,
):
DbJoinModelABC.__init__(self, source_id, reference_id, id, deleted, editor_id, created, updated)
self._source_id = source_id
self._reference_id = reference_id
@property
def source_id(self) -> int:
return self._source_id
@property
def reference(self) -> int:
return self._reference_id

View File

@@ -0,0 +1,5 @@
from enum import Enum
class <Name>Enum(Enum):
KEY = "value"

View File

@@ -0,0 +1,7 @@
from cpl.core.log.wrapped_logger import WrappedLogger
class <Name>Logger(WrappedLogger):
def __init__(self):
WrappedLogger.__init__(self, "<name>")

View File

@@ -0,0 +1,17 @@
from cpl.dependency import ServiceCollection, ServiceProvider
from cpl.dependency.module import Module
class <Name>Module(Module):
dependencies = []
configuration = []
singleton = []
scoped = []
transient = []
hosted = []
@staticmethod
def register(collection: ServiceCollection): ...
@staticmethod
def configure(provider: ServiceProvider): ...

View File

@@ -0,0 +1,9 @@
import multiprocessing
class <Name>(multiprocessing.Process):
def __init__(self):
multiprocessing.Process.__init__(self)
def run(self): ...

View File

@@ -0,0 +1,11 @@
from cpl.core.pipes import PipeABC
from cpl.core.typing import T
class <Name>Pipe(PipeABC):
@staticmethod
def to_str(value: T, *args) -> str: ...
@staticmethod
def from_str(value: str, *args) -> T: ...

View File

@@ -0,0 +1,9 @@
import threading
class <Name>(threading.Thread):
def __init__(self):
threading.Thread.__init__(self)
def run(self): ...

View File

@@ -0,0 +1,18 @@
from cpl.api.application import WebApp
from cpl.core.environment import Environment
from cpl.core.log import LoggerABC
from cpl.dependency import ServiceProvider
from cpl.dependency.typing import Modules
class <Name>(WebApp):
def __init__(self, services: ServiceProvider, modules: Modules):
WebApp.__init__(self, services, modules)
self._logger = services.get_service(LoggerABC)
async def main(self):
self._logger.debug(f"Host: {Environment.get_host_name()}")
self._logger.debug(f"Environment: {Environment.get_environment()}")
await super().main()

View File

@@ -0,0 +1,122 @@
import os
import re
from pathlib import Path
import click
from cpl.cli import cli as clim
from cpl.cli.cli import cli
from cpl.cli.model.project import Project
from cpl.cli.model.workspace import Workspace
from cpl.cli.utils.structure import get_project_by_name_or_path
from cpl.cli.utils.template_collector import TemplateCollector
from cpl.core.configuration import Configuration
from cpl.core.console import Console
from cpl.core.utils import String
@cli.command("generate", aliases=["g"])
@click.argument("schematic", type=click.STRING, required=True)
@click.argument("name", type=click.STRING, required=True)
@click.option("--verbose", "-v", is_flag=True, help="Enable verbose output")
def generate(schematic: str, name: str, verbose: bool) -> None:
TemplateCollector.collect_templates(Path(clim.__file__).parent, verbose)
workspace: Workspace = Configuration.get("workspace")
if workspace is not None:
if verbose:
Console.write_line("Workspace found, collecting templates...")
TemplateCollector.collect_templates(Path(workspace.path).parent, verbose)
project = None
try:
project = get_project_by_name_or_path("./")
if verbose:
Console.write_line("project found, collecting templates...")
TemplateCollector.collect_templates(Path(project.path).parent, verbose)
except ValueError:
if verbose:
Console.write_line("Local project not found")
templates = TemplateCollector.get_templates()
schematics = {}
for template_name, template_content in templates.items():
t_name = template_name.split(".")[0]
if t_name in schematics:
raise ValueError(f"Duplicate schematic name found: {t_name}")
schematics[t_name] = template_name
for i in range(len(t_name)):
char = t_name[i]
if char in schematics:
continue
schematics[char] = template_name
break
if schematic not in schematics:
raise ValueError(
f"Schematic '{schematic}' not found. Available schematics: {', '.join([x.split(".")[0] for x in templates.keys()])}"
)
path, name = _get_name_and_path_from_name(name, project)
os.makedirs(path, exist_ok=True)
Console.write_line(f"Generating {str(path / name)} ...")
with open(path / f"{name}.py", "w") as f:
f.write(
_render_template(schematics[schematic].split(".")[0], templates[schematics[schematic]], name, str(path))
)
def _get_name_and_path_from_name(in_name: str, project: Project = None) -> tuple[Path, str]:
path = ""
name = ""
in_name_parts = in_name.split("/")
if len(in_name_parts) == 1:
name = in_name_parts[0]
else:
path = "/".join(in_name_parts[:-1])
name = in_name_parts[-1]
workspace: Workspace = Configuration.get("workspace")
if workspace is None and project is not None:
return (Path(project.path).parent / project.directory / path).resolve().absolute(), name
elif workspace is None and project is None:
return Path(path).resolve().absolute(), name
selected_project = project
project_name = path.split("/")[0]
project_by_name = workspace.get_project_by_name(project_name)
if project_by_name is not None:
selected_project = project_by_name
path = "/".join(path.split("/")[1:])
if selected_project is None:
selected_project = workspace.get_project_by_name(workspace.default_project)
return (Path(selected_project.path).parent / selected_project.directory / path).resolve().absolute(), name
def _render_template(schematic, template_str: str, name: str, path: str) -> str:
context = {
"schematic": schematic,
"Name": String.to_pascal_case(name),
"name": String.to_snake_case(name),
"NAME": String.to_snake_case(name).upper(),
"camelName": String.to_camel_case(name),
"multi_Name": f"{String.to_pascal_case(name)}s",
"multi_name": f"{String.to_snake_case(name)}s",
"multi_NAME": f"{String.to_snake_case(name).upper()}s",
"multi_camelName": f"{String.to_camel_case(name)}s",
"path": path.replace("\\", "/"),
}
for key, value in context.items():
template_str = template_str.replace(f"<{key}>", value)
return template_str

View File

@@ -10,6 +10,7 @@ from cpl.cli.command.package.update import update
from cpl.cli.command.project.build import build
from cpl.cli.command.project.run import run
from cpl.cli.command.project.start import start
from cpl.cli.command.structure.generate import generate
from cpl.cli.command.structure.init import init
from cpl.cli.command.version import version
from cpl.cli.model.workspace import Workspace
@@ -54,6 +55,7 @@ def configure():
# structure
cli.add_command(init)
# cli.add_command(new)
cli.add_command(generate)
# packaging
cli.add_command(install)

View File

@@ -50,6 +50,10 @@ class Workspace(CPLStructureModel):
def actual_projects(self) -> List[Project]:
return [Project.from_file(p) for p in self._projects]
@property
def project_names(self) -> List[str]:
return [Project.from_file(p).name for p in self._projects if "name" in p]
@property
def default_project(self) -> Optional[str]:
return self._default_project
@@ -65,3 +69,9 @@ class Workspace(CPLStructureModel):
@scripts.setter
def scripts(self, value: Dict[str, str]):
self._scripts = self._require_dict_str_str(value, "scripts")
def get_project_by_name(self, name: str) -> Optional[Project]:
for project in self.actual_projects:
if project.name == name:
return project
return None

View File

@@ -0,0 +1,13 @@
class NameUtils:
@staticmethod
def classify(name: str) -> str: # UserService
return "".join(w.capitalize() for w in name.replace("-", "_").split("_"))
@staticmethod
def dasherize(name: str) -> str: # user-service
return name.replace("_", "-").lower()
@staticmethod
def camelize(name: str) -> str: # userService
parts = name.split("-")
return parts[0] + "".join(w.capitalize() for w in parts[1:])

View File

@@ -0,0 +1,41 @@
import os
from pathlib import Path
from cpl.core.console import Console
class TemplateCollector:
_templates = {}
_collected_paths = []
@classmethod
def get_templates(cls) -> dict[str, str]:
return cls._templates
@classmethod
def collect_templates(cls, directory: Path, verbose=False):
if not directory.exists() or not directory.is_dir():
raise FileNotFoundError(f"Directory '{directory}' does not exist")
if not str(directory).endswith(".cpl/generate"):
directory = directory / ".cpl" / "generate"
directory = directory.resolve().absolute()
if directory in cls._collected_paths:
return
cls._collected_paths.append(directory)
if not directory.exists() or not directory.is_dir():
if verbose:
Console.write_line(f"No templates found in {directory}")
return
templates = {}
for root, _, files in os.walk(directory):
for file in files:
if file.endswith(".schematic"):
with open(os.path.join(root, file), "r") as f:
templates[os.path.relpath(os.path.join(root, file), directory)] = f.read()
cls._templates.update(templates)

View File

@@ -0,0 +1 @@
from .module import Module