Added enumerable order & added array & removed collection

Reviewed-on: #193

.gitea/workflows/build-dev.yaml

@@ -12,6 +12,27 @@ jobs:
       version_suffix: 'dev'
     secrets: inherit
 
+  api:
+    uses: ./.gitea/workflows/package.yaml
+    needs: [ prepare, application, auth, core, dependency ]
+    with:
+      working_directory: src/cpl-api
+    secrets: inherit
+
+  application:
+    uses: ./.gitea/workflows/package.yaml
+    needs: [ prepare, core, dependency ]
+    with:
+      working_directory: src/cpl-application
+    secrets: inherit
+
+  auth:
+    uses: ./.gitea/workflows/package.yaml
+    needs: [ prepare, core, dependency,  database ]
+    with:
+      working_directory: src/cpl-auth
+    secrets: inherit
+
   core:
     uses: ./.gitea/workflows/package.yaml
     needs: [prepare]
@@ -19,6 +40,27 @@ jobs:
       working_directory: src/cpl-core
     secrets: inherit
 
+  database:
+    uses: ./.gitea/workflows/package.yaml
+    needs: [ prepare, core, dependency ]
+    with:
+      working_directory: src/cpl-database
+    secrets: inherit
+
+  dependency:
+    uses: ./.gitea/workflows/package.yaml
+    needs: [ prepare, core ]
+    with:
+      working_directory: src/cpl-dependency
+    secrets: inherit
+
+  mail:
+    uses: ./.gitea/workflows/package.yaml
+    needs: [ prepare, core, dependency ]
+    with:
+      working_directory: src/cpl-mail
+    secrets: inherit
+
   query:
     uses: ./.gitea/workflows/package.yaml
     needs: [prepare]
@@ -28,14 +70,7 @@ jobs:
 
   translation:
     uses: ./.gitea/workflows/package.yaml
-    needs: [ prepare, core ]
+    needs: [ prepare, core, dependency ]
     with:
       working_directory: src/cpl-translation
-    secrets: inherit
-
-  mail:
-    uses: ./.gitea/workflows/package.yaml
-    needs: [ prepare, core ]
-    with:
-      working_directory: src/cpl-mail
     secrets: inherit

.gitea/workflows/test_before_merge.yaml

@@ -0,0 +1,26 @@
+name: Test before pr merge
+run-name: Test before pr merge
+on:
+  pull_request:
+    types:
+      - opened
+      - edited
+      - reopened
+      - synchronize
+      - ready_for_review
+
+jobs:
+  test-lint:
+    runs-on: [ runner ]
+    container: git.sh-edraft.de/sh-edraft.de/act-runner:latest
+    steps:
+      - name: Clone Repository
+        uses: https://github.com/actions/checkout@v3
+        with:
+          token: ${{ secrets.CI_ACCESS_TOKEN }}
+
+      - name: Installing black
+        run: python3.12 -m pip install black
+
+      - name: Checking black
+        run: python3.12 -m black src --check

.gitignore

@@ -113,6 +113,7 @@ venv.bak/
 
 # Custom Environments
 cpl-env/
+.secret
 
 # Spyder project settings
 .spyderproject
@@ -138,3 +139,6 @@ PythonImportHelper-v2-Completion.json
 
 # cpl unittest stuff
 unittests/test_*_playground
+
+# cpl logs
+**/logs/*.jsonl

README.md

@@ -1,153 +0,0 @@
-<h1 align="center">CPL - Common python library</h1>
-
-<!-- Summary -->
-<p align="center">
-  <!-- <img src="" alt="cpl-logo" width="120px" height="120px"/> -->
-  <br>
-  <i>
-  CPL is a development platform for python server applications
-  <br>using Python.</i>
-  <br>
-</p>
-
-## Table of Contents
-<!-- TABLE OF CONTENTS -->
-<ol>
-   <li><a href="#Features">Features</a></li>
-   <li>
-     <a href="#getting-started">Getting Started</a>
-     <ul>
-       <li><a href="#prerequisites">Prerequisites</a></li>
-       <li><a href="#installation">Installation</a></li>
-     </ul>
-   </li>
-   <li><a href="#roadmap">Roadmap</a></li>
-   <li><a href="#contributing">Contributing</a></li>
-   <li><a href="#license">License</a></li>
-   <li><a href="#contact">Contact</a></li>
-</ol>
-
-## Features
-<!-- FEATURE OVERVIEW -->
-- Expandle
-- Application base
-    - Standardized application classes
-    - Application object builder
-    - Application extension classes
-    - Startup classes
-    - Startup extension classes
-- Configuration
-    - Configure via object mapped JSON
-    - Console argument handling
-- Console class for in and output
-    - Banner
-    - Spinner
-    - Options (menu)
-    - Table
-    - Write
-    - Write_at
-    - Write_line
-    - Write_line_at
-- Dependency injection
-    - Service lifetimes: singleton, scoped and transient
-- Providing of application environment
-    - Environment (development, staging, testing, production)
-    - Appname
-    - Customer
-    - Hostname
-    - Runtime directory
-    - Working directory
-- Logging
-    - Standardized logger
-    - Log-level (FATAL, ERROR, WARN, INFO, DEBUG & TRACE)
-- Mail handling
-    - Send mails
-- Pipe classes
-    - Convert input
-- Utils
-    - Credential manager
-        - Encryption via BASE64
-    - PIP wrapper class based on subprocess
-        - Run pip commands
-    - String converter to different variants
-        - to_lower_case
-        - to_camel_case
-        - ...
-
-<!-- GETTING STARTED -->
-## Getting Started
-
-[Get started with CPL][quickstart].
-
-### Prerequisites
-
-- Install [python] which includes [Pip installs packages][pip]
-
-### Installation
-
-Install the CPL package
-```sh
-pip install cpl-core --extra-index-url https://pip.sh-edraft.de
-```
-
-Install the CPL CLI
-```sh
-pip install cpl-cli --extra-index-url https://pip.sh-edraft.de
-```
-
-Create workspace:
-```sh
-cpl new <console|library|unittest> <PROJECT NAME>
-```
-
-Run the application:
-```sh
-cd <PROJECT NAME>
-cpl start
-```
-
-
-<!-- ROADMAP -->
-## Roadmap
-
-See the [open issues](https://git.sh-edraft.de/sh-edraft.de/sh_cpl/issues) for a list of proposed features (and known issues).
-
-
-
-<!-- CONTRIBUTING -->
-## Contributing
-
-### Contributing Guidelines
-
-Read through our [contributing guidelines][contributing] to learn about our submission process, coding rules and more.
-
-### Want to Help?
-
-Want to file a bug, contribute some code, or improve documentation? Excellent! Read up on our guidelines for [contributing][contributing].
-
-
-
-<!-- LICENSE -->
-## License
-
-Distributed under the MIT License. See [LICENSE] for more information.
-
-
-
-<!-- CONTACT -->
-## Contact
-
-Sven Heidemann - sven.heidemann@sh-edraft.de
-
-Project link: [https://git.sh-edraft.de/sh-edraft.de/sh_common_py_lib](https://git.sh-edraft.de/sh-edraft.de/sh_cpl)
-
-<!-- External LINKS -->
-[pip_url]: https://pip.sh-edraft.de
-[python]: https://www.python.org/
-[pip]: https://pypi.org/project/pip/
-
-<!-- Internal LINKS -->
-[project]: https://git.sh-edraft.de/sh-edraft.de/sh_cpl
-[quickstart]: https://git.sh-edraft.de/sh-edraft.de/sh_cpl/wiki/quickstart
-[contributing]: https://git.sh-edraft.de/sh-edraft.de/sh_cpl/wiki/contributing
-[license]: LICENSE

example/custom/api/src/appsettings.development.json

@@ -0,0 +1,8 @@
+{
+  "Logging": {
+    "Path": "logs/",
+    "Filename": "log_$start_time.log",
+    "ConsoleLevel": "TRACE",
+    "Level": "TRACE"
+  }
+}

example/custom/api/src/appsettings.edrafts-pc.json

@@ -1,23 +1,24 @@
 {
-  "TimeFormatSettings": {
+  "TimeFormat": {
     "DateFormat": "%Y-%m-%d",
     "TimeFormat": "%H:%M:%S",
     "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
     "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
   },
 
-  "LoggingSettings": {
+  "Log": {
     "Path": "logs/",
     "Filename": "log_$start_time.log",
-    "ConsoleLogLevel": "TRACE",
-    "FileLogLevel": "TRACE"
+    "ConsoleLevel": "TRACE",
+    "Level": "TRACE"
   },
 
-  "DatabaseSettings": {
+  "Database": {
     "Host": "localhost",
-    "User": "sh_cpl",
-    "Password": "MHZhc0Y2bjhKc1VUMWV0Qw==",
-    "Database": "sh_cpl",
+    "User": "cpl",
+    "Port": 3306,
+    "Password": "cpl",
+    "Database": "cpl",
     "Charset": "utf8mb4",
     "UseUnicode": "true",
     "Buffered": "true"

example/custom/api/src/appsettings.json

@@ -1,15 +1,15 @@
 {
-  "TimeFormatSettings": {
+  "TimeFormat": {
     "DateFormat": "%Y-%m-%d",
     "TimeFormat": "%H:%M:%S",
     "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
     "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
   },
 
-  "LoggingSettings": {
+  "Log": {
     "Path": "logs/",
     "Filename": "log_$start_time.log",
-    "ConsoleLogLevel": "ERROR",
-    "FileLogLevel": "WARN"
+    "ConsoleLevel": "ERROR",
+    "Level": "WARNING"
   }
 }

example/custom/api/src/main.py

@@ -0,0 +1,47 @@
+from starlette.responses import JSONResponse
+
+from cpl import api
+from cpl.api.application.web_app import WebApp
+from cpl.application import ApplicationBuilder
+from cpl.auth.permission.permissions import Permissions
+from cpl.auth.schema import AuthUser, Role
+from cpl.core.configuration import Configuration
+from cpl.core.environment import Environment
+from cpl.core.utils.cache import Cache
+from service import PingService
+
+
+def main():
+    builder = ApplicationBuilder[WebApp](WebApp)
+
+    Configuration.add_json_file(f"appsettings.json")
+    Configuration.add_json_file(f"appsettings.{Environment.get_environment()}.json")
+    Configuration.add_json_file(f"appsettings.{Environment.get_host_name()}.json", optional=True)
+
+    # builder.services.add_logging()
+    builder.services.add_structured_logging()
+    builder.services.add_transient(PingService)
+    builder.services.add_module(api)
+
+    builder.services.add_cache(AuthUser)
+    builder.services.add_cache(Role)
+
+    app = builder.build()
+    app.with_logging()
+    app.with_database()
+
+    app.with_authentication()
+    app.with_authorization()
+
+    app.with_route(path="/route1", fn=lambda r: JSONResponse("route1"), method="GET", authentication=True, permissions=[Permissions.administrator])
+    app.with_routes_directory("routes")
+
+    provider = builder.service_provider
+    user_cache = provider.get_service(Cache[AuthUser])
+    role_cache = provider.get_service(Cache[Role])
+
+    app.run()
+
+
+if __name__ == "__main__":
+    main()

example/custom/api/src/routes/ping.py

@@ -0,0 +1,16 @@
+from urllib.request import Request
+
+from service import PingService
+from starlette.responses import JSONResponse
+
+from cpl.api import APILogger
+from cpl.api.router import Router
+
+
+@Router.authenticate()
+# @Router.authorize(permissions=[Permissions.administrator])
+# @Router.authorize(policies=["test"])
+@Router.get(f"/ping")
+async def ping(r: Request, ping: PingService, logger: APILogger):
+    logger.info(f"Ping: {ping}")
+    return JSONResponse(ping.ping(r))

example/custom/api/src/service.py

@@ -0,0 +1,4 @@
+class PingService:
+
+    def ping(self, r):
+        return "pong"

example/custom/console/main.py

@@ -15,32 +15,27 @@ def test_console():
     Console.write_line("Hello World")
     Console.write("\nName: ")
     Console.write_line(" Hello", Console.read_line())
-    Console.clear()
     Console.write_at(5, 5, "at 5, 5")
     Console.write_at(10, 10, "at 10, 10")
 
 
 if __name__ == "__main__":
     Console.write_line("Hello World\n")
+    Console.clear()
     Console.spinner(
         "Test:", test_spinner, spinner_foreground_color=ForegroundColorEnum.cyan, text_foreground_color="green"
     )
+    test_console()
     Console.write_line("HOLD BACK")
-    # opts = [
-    #     'Option 1',
-    #     'Option 2',
-    #     'Option 3',
-    #     'Option 4'
-    # ]
-    # selected = Console.select(
-    #     '>',
-    #     'Select item:',
-    #     opts,
-    #     header_foreground_color=ForegroundColorEnum.blue,
-    #     option_foreground_color=ForegroundColorEnum.green,
-    #     cursor_foreground_color=ForegroundColorEnum.red
-    # )
-    # Console.write_line(f'You selected: {selected}')
-    # # test_console()
-    #
-    # Console.write_line()
+    opts = ["Option 1", "Option 2", "Option 3", "Option 4"]
+    selected = Console.select(
+        ">",
+        "Select item:",
+        opts,
+        header_foreground_color=ForegroundColorEnum.blue,
+        option_foreground_color=ForegroundColorEnum.green,
+        cursor_foreground_color=ForegroundColorEnum.red,
+    )
+    Console.write_line(f"You selected: {selected}")
+
+    Console.write_line()

example/custom/database/cpl.json

@@ -1,5 +1,5 @@
 {
-  "ProjectSettings": {
+  "Project": {
     "Name": "database",
     "Version": {
       "Major": "0",
@@ -22,7 +22,7 @@
     "PythonPath": {},
     "Classifiers": []
   },
-  "BuildSettings": {
+  "Build": {
     "ProjectType": "console",
     "SourcePath": "src",
     "OutputPath": "dist",

example/custom/database/src/application.py

@@ -0,0 +1,39 @@
+from cpl.application.abc import ApplicationABC
+from cpl.auth.keycloak import KeycloakAdmin
+from cpl.core.console import Console
+from cpl.core.environment import Environment
+from cpl.core.log import LoggerABC
+from cpl.dependency import ServiceProviderABC
+from model.city import City
+from model.city_dao import CityDao
+from model.user import User
+from model.user_dao import UserDao
+
+
+class Application(ApplicationABC):
+    def __init__(self, services: ServiceProviderABC):
+        ApplicationABC.__init__(self, services)
+
+        self._logger = services.get_service(LoggerABC)
+
+    async def test_daos(self):
+        userDao: UserDao = self._services.get_service(UserDao)
+        cityDao: CityDao = self._services.get_service(CityDao)
+
+        Console.write_line(await userDao.get_all())
+
+        if len(await cityDao.get_all()) == 0:
+            city_id = await cityDao.create(City(0, "Haren", "49733"))
+            await userDao.create(User(0, "NewUser", city_id))
+
+        Console.write_line(await userDao.get_all())
+
+    async def main(self):
+        self._logger.debug(f"Host: {Environment.get_host_name()}")
+        self._logger.debug(f"Environment: {Environment.get_environment()}")
+
+        await self.test_daos()
+
+        kc_admin: KeycloakAdmin = self._services.get_service(KeycloakAdmin)
+        x = kc_admin.get_users()
+        Console.write_line(x)

example/custom/database/src/appsettings.development.json

@@ -0,0 +1,8 @@
+{
+  "Logging": {
+    "Path": "logs/",
+    "Filename": "log_$start_time.log",
+    "ConsoleLevel": "TRACE",
+    "Level": "TRACE"
+  }
+}

example/custom/database/src/appsettings.edrafts-lapi.json

@@ -1,22 +1,22 @@
 {
-  "TimeFormatSettings": {
+  "TimeFormat": {
     "DateFormat": "%Y-%m-%d",
     "TimeFormat": "%H:%M:%S",
     "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
     "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
   },
 
-  "LoggingSettings": {
+  "Log": {
     "Path": "logs/",
     "Filename": "log_$start_time.log",
-    "ConsoleLogLevel": "TRACE",
-    "FileLogLevel": "TRACE"
+    "ConsoleLevel": "TRACE",
+    "Level": "TRACE"
   },
 
-  "DatabaseSettings": {
+  "Database": {
     "AuthPlugin": "mysql_native_password",
-    "ConnectionString": "mysql+mysqlconnector://sh_cpl:$credentials@localhost/sh_cpl",
-    "Credentials": "MHZhc0Y2bjhKc1VUMWV0Qw==",
+    "ConnectionString": "mysql+mysqlconnector://cpl:$credentials@localhost/cpl",
+    "Credentials": "Y3Bs",
     "Encoding": "utf8mb4"
   }
 }

example/custom/database/src/appsettings.edrafts-pc.json

@@ -0,0 +1,26 @@
+{
+  "TimeFormat": {
+    "DateFormat": "%Y-%m-%d",
+    "TimeFormat": "%H:%M:%S",
+    "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
+    "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
+  },
+
+  "Log": {
+    "Path": "logs/",
+    "Filename": "log_$start_time.log",
+    "ConsoleLevel": "TRACE",
+    "Level": "TRACE"
+  },
+
+  "Database": {
+    "Host": "localhost",
+    "User": "cpl",
+    "Port": 3306,
+    "Password": "cpl",
+    "Database": "cpl",
+    "Charset": "utf8mb4",
+    "UseUnicode": "true",
+    "Buffered": "true"
+  }
+}

example/custom/database/src/appsettings.json

@@ -1,15 +1,15 @@
 {
-  "TimeFormatSettings": {
+  "TimeFormat": {
     "DateFormat": "%Y-%m-%d",
     "TimeFormat": "%H:%M:%S",
     "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
     "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
   },
 
-  "LoggingSettings": {
+  "Log": {
     "Path": "logs/",
     "Filename": "log_$start_time.log",
-    "ConsoleLogLevel": "ERROR",
-    "FileLogLevel": "WARN"
+    "ConsoleLevel": "ERROR",
+    "Level": "WARNING"
   }
 }

example/custom/database/src/custom_permissions.py

@@ -0,0 +1,5 @@
+from enum import Enum
+
+
+class CustomPermissions(Enum):
+    test = "test"

example/custom/database/src/main.py

@@ -0,0 +1,19 @@
+from application import Application
+from cpl.application import ApplicationBuilder
+from custom_permissions import CustomPermissions
+from startup import Startup
+
+
+def main():
+    builder = ApplicationBuilder(Application).with_startup(Startup)
+    app = builder.build()
+
+    app.with_logging()
+    app.with_permissions(CustomPermissions)
+    app.with_migrations("./scripts")
+    app.with_seeders()
+    app.run()
+
+
+if __name__ == "__main__":
+    main()

example/custom/database/src/main_simplified.py

@@ -0,0 +1,27 @@
+from application import Application
+from cpl.application import ApplicationBuilder
+from cpl.auth.permission.permissions_registry import PermissionsRegistry
+from cpl.core.console import Console
+from cpl.core.log import LogLevel
+from custom_permissions import CustomPermissions
+from startup import Startup
+
+
+def main():
+    builder = ApplicationBuilder(Application).with_startup(Startup)
+    builder.services.add_logging()
+
+    app = builder.build()
+
+    app.with_logging(LogLevel.trace)
+    app.with_permissions(CustomPermissions)
+    app.with_migrations("./scripts")
+    app.with_seeders()
+
+    Console.write_line(CustomPermissions.test.value in PermissionsRegistry.get())
+    app.run()
+    Console.write_line("Hello from main_simplified.py!")
+
+
+if __name__ == "__main__":
+    main()

example/custom/database/src/model/city.py

@@ -0,0 +1,29 @@
+from datetime import datetime
+from typing import Optional
+
+from cpl.core.typing import SerialId
+from cpl.database.abc.db_model_abc import DbModelABC
+
+
+class City(DbModelABC):
+    def __init__(
+        self,
+        id: int,
+        name: str,
+        zip: str,
+        deleted: bool = False,
+        editor_id: Optional[SerialId] = None,
+        created: Optional[datetime] = None,
+        updated: Optional[datetime] = None,
+    ):
+        DbModelABC.__init__(self, id, deleted, editor_id, created, updated)
+        self._name = name
+        self._zip = zip
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def zip(self) -> str:
+        return self._zip

example/custom/database/src/model/city_dao.py

@@ -0,0 +1,11 @@
+from cpl.database.abc import DbModelDaoABC
+from model.city import City
+
+
+class CityDao(DbModelDaoABC[City]):
+
+    def __init__(self):
+        DbModelDaoABC.__init__(self, City, "city")
+
+        self.attribute(City.name, str)
+        self.attribute(City.zip, int)

example/custom/database/src/model/user.py

@@ -0,0 +1,30 @@
+from datetime import datetime
+from typing import Optional
+
+from cpl.core.typing import SerialId
+from cpl.database.abc.db_model_abc import DbModelABC
+
+
+class User(DbModelABC):
+
+    def __init__(
+        self,
+        id: int,
+        name: str,
+        city_id: int = 0,
+        deleted: bool = False,
+        editor_id: Optional[SerialId] = None,
+        created: Optional[datetime] = None,
+        updated: Optional[datetime] = None,
+    ):
+        DbModelABC.__init__(self, id, deleted, editor_id, created, updated)
+        self._name = name
+        self._city_id = city_id
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def city_id(self) -> int:
+        return self._city_id

example/custom/database/src/model/user_dao.py

@@ -0,0 +1,13 @@
+from cpl.database.abc import DbModelDaoABC
+from model.user import User
+
+
+class UserDao(DbModelDaoABC[User]):
+
+    def __init__(self):
+        DbModelDaoABC.__init__(self, User, "users")
+
+        self.attribute(User.name, str)
+        self.attribute(User.city_id, int, db_name="CityId")
+
+        self.reference("city", "id", User.city_id, "city")

example/custom/database/src/scripts/0-initial.sql

@@ -0,0 +1,22 @@
+CREATE TABLE IF NOT EXISTS `city` (
+    `id` INT(30) NOT NULL AUTO_INCREMENT,
+    `name` VARCHAR(64) NOT NULL,
+    `zip` VARCHAR(5) NOT NULL,
+    deleted BOOLEAN NOT NULL DEFAULT FALSE,
+    editorId INT NULL,
+    created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    PRIMARY KEY(`id`)
+);
+
+CREATE TABLE IF NOT EXISTS `users` (
+    `id` INT(30) NOT NULL AUTO_INCREMENT,
+    `name` VARCHAR(64) NOT NULL,
+    `cityId` INT(30),
+    deleted BOOLEAN NOT NULL DEFAULT FALSE,
+    editorId INT NULL,
+    created TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP,
+    updated TIMESTAMP NOT NULL DEFAULT CURRENT_TIMESTAMP ON UPDATE CURRENT_TIMESTAMP,
+    FOREIGN KEY (`cityId`) REFERENCES city(`id`),
+    PRIMARY KEY(`id`)
+);

example/custom/database/src/startup.py

@@ -0,0 +1,31 @@
+from cpl import auth
+from cpl.application.abc.startup_abc import StartupABC
+from cpl.auth import permission
+from cpl.core.configuration import Configuration
+from cpl.core.environment import Environment
+from cpl.core.log import Logger, LoggerABC
+from cpl.database import mysql
+from cpl.database.abc.data_access_object_abc import DataAccessObjectABC
+from cpl.dependency import ServiceCollection
+from model.city_dao import CityDao
+from model.user_dao import UserDao
+
+
+class Startup(StartupABC):
+
+    @staticmethod
+    async def configure_configuration():
+        Configuration.add_json_file(f"appsettings.json")
+        Configuration.add_json_file(f"appsettings.{Environment.get_environment()}.json")
+        Configuration.add_json_file(f"appsettings.{Environment.get_host_name()}.json", optional=True)
+
+    @staticmethod
+    async def configure_services(services: ServiceCollection):
+        services.add_module(mysql)
+        services.add_module(auth)
+        services.add_module(permission)
+
+        services.add_transient(DataAccessObjectABC, UserDao)
+        services.add_transient(DataAccessObjectABC, CityDao)
+
+        services.add_singleton(LoggerABC, Logger)

example/custom/di/appsettings.json

@@ -1,15 +1,15 @@
 {
-  "TimeFormatSettings": {
+  "TimeFormat": {
     "DateFormat": "%Y-%m-%d",
     "TimeFormat": "%H:%M:%S",
     "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
     "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
   },
 
-  "LoggingSettings": {
+  "Logging": {
     "Path": "logs/",
     "Filename": "log_$start_time.log",
-    "ConsoleLogLevel": "ERROR",
-    "FileLogLevel": "WARN"
+    "ConsoleLevel": "ERROR",
+    "Level": "WARN"
   }
 }

example/custom/di/cpl-workspace.json

@@ -1,5 +1,5 @@
 {
-  "WorkspaceSettings": {
+  "Workspace": {
     "DefaultProject": "di",
     "Projects": {
       "di": "src/di/di.json"

example/custom/di/src/di/application.py

@@ -1,5 +1,4 @@
-from cpl.application import ApplicationABC
-from cpl.core.configuration import ConfigurationABC
+from cpl.application.abc import ApplicationABC
 from cpl.core.console.console import Console
 from cpl.dependency import ServiceProviderABC
 from cpl.dependency.scope import Scope
@@ -11,15 +10,14 @@ from di.tester import Tester
 
 
 class Application(ApplicationABC):
-    def __init__(self, config: ConfigurationABC, services: ServiceProviderABC):
-        ApplicationABC.__init__(self, config, services)
+    def __init__(self, services: ServiceProviderABC):
+        ApplicationABC.__init__(self, services)
 
     def _part_of_scoped(self):
         ts: TestService = self._services.get_service(TestService)
         ts.run()
 
-    def configure(self):
-        pass
+    def configure(self): ...
 
     def main(self):
         with self._services.create_scope() as scope:

example/custom/di/src/di/di.json

@@ -1,5 +1,5 @@
 {
-  "ProjectSettings": {
+  "Project": {
     "Name": "di",
     "Version": {
       "Major": "0",
@@ -25,7 +25,7 @@
     "PythonPath": {},
     "Classifiers": []
   },
-  "BuildSettings": {
+  "Build": {
     "ProjectType": "console",
     "SourcePath": "",
     "OutputPath": "../../dist",

example/custom/di/src/di/main.py

@@ -6,7 +6,7 @@ from di.startup import Startup
 
 def main():
     app_builder = ApplicationBuilder(Application)
-    app_builder.use_startup(Startup)
+    app_builder.with_startup(Startup)
     app_builder.build().run()
 
 

example/custom/di/src/di/startup.py

@@ -1,12 +1,10 @@
-from cpl.application import StartupABC
-from cpl.core.configuration import ConfigurationABC
+from cpl.application.abc import StartupABC
 from cpl.dependency import ServiceProviderABC, ServiceCollection
-from cpl.core.environment import Environment
+from di.di_tester_service import DITesterService
 from di.test1_service import Test1Service
 from di.test2_service import Test2Service
 from di.test_abc import TestABC
 from di.test_service import TestService
-from di.di_tester_service import DITesterService
 from di.tester import Tester
 
 
@@ -14,10 +12,9 @@ class Startup(StartupABC):
     def __init__(self):
         StartupABC.__init__(self)
 
-    def configure_configuration(self, configuration: ConfigurationABC, environment: Environment) -> ConfigurationABC:
-        return configuration
+    def configure_configuration(self): ...
 
-    def configure_services(self, services: ServiceCollection, environment: Environment) -> ServiceProviderABC:
+    def configure_services(self, services: ServiceCollection) -> ServiceProviderABC:
         services.add_scoped(TestService)
         services.add_scoped(DITesterService)
 
@@ -25,4 +22,4 @@ class Startup(StartupABC):
         services.add_singleton(TestABC, Test2Service)
         services.add_singleton(Tester)
 
-        return services.build_service_provider()
+        return services.build()

example/custom/di/src/di/static_test.py

@@ -1,4 +1,3 @@
-from cpl.core.configuration import ConfigurationABC
 from cpl.dependency import ServiceProvider, ServiceProviderABC
 from di.test_service import TestService
 
@@ -6,5 +5,5 @@ from di.test_service import TestService
 class StaticTest:
     @staticmethod
     @ServiceProvider.inject
-    def test(services: ServiceProviderABC, config: ConfigurationABC, t1: TestService):
+    def test(services: ServiceProviderABC, t1: TestService):
         t1.run()

example/custom/di/src/di/test_service.py

@@ -6,7 +6,7 @@ from cpl.core.utils.string import String
 
 class TestService:
     def __init__(self):
-        self._name = String.random_string(string.ascii_lowercase, 8)
+        self._name = String.random(8)
 
     def run(self):
         Console.write_line(f"Im {self._name}")

example/custom/general/src/general/application.py

@@ -1,7 +1,7 @@
 import time
 from typing import Optional
 
-from cpl.application.application_abc import ApplicationABC
+from cpl.application.abc import ApplicationABC
 from cpl.core.configuration import Configuration
 from cpl.core.console import Console
 from cpl.dependency import ServiceProviderABC
@@ -18,8 +18,8 @@ class Application(ApplicationABC):
 
     def __init__(self, services: ServiceProviderABC):
         ApplicationABC.__init__(self, services)
-        self._logger: Optional[LoggerABC] = None
-        self._mailer: Optional[EMailClientABC] = None
+        self._logger = self._services.get_service(LoggerABC)
+        self._mailer = self._services.get_service(EMailClientABC)
 
     def test_send_mail(self):
         mail = EMail()
@@ -35,10 +35,6 @@ class Application(ApplicationABC):
     def _wait(time_ms: int):
         time.sleep(time_ms)
 
-    def configure(self):
-        self._logger = self._services.get_service(LoggerABC)
-        self._mailer = self._services.get_service(EMailClientABC)
-
     def main(self):
         self._logger.debug(f"Host: {Environment.get_host_name()}")
         self._logger.debug(f"Environment: {Environment.get_environment()}")

example/custom/general/src/general/appsettings.development.json

@@ -0,0 +1,8 @@
+{
+  "Logging": {
+    "Path": "logs/",
+    "Filename": "log_$start_time.log",
+    "ConsoleLevel": "TRACE",
+    "Level": "TRACE"
+  }
+}

example/custom/general/src/general/appsettings.edrafts-lapi.json

@@ -0,0 +1,20 @@
+{
+  "TimeFormat": {
+    "DateFormat": "%Y-%m-%d",
+    "TimeFormat": "%H:%M:%S",
+    "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
+    "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
+  },
+  "Logging": {
+    "Path": "logs/",
+    "Filename": "log_$start_time.log",
+    "ConsoleLevel": "TRACE",
+    "Level": "TRACE"
+  },
+  "EMailClient": {
+    "Host": "mail.sh-edraft.de",
+    "Port": "587",
+    "UserName": "dev-srv@sh-edraft.de",
+    "Credentials": "RmBOQX1eNFYiYjgsSid3fV1nelc2WA=="
+  }
+}

example/custom/general/src/general/appsettings.edrafts-pc.json

@@ -1,26 +1,26 @@
 {
-  "TimeFormatSettings": {
+  "TimeFormat": {
     "DateFormat": "%Y-%m-%d",
     "TimeFormat": "%H:%M:%S",
     "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
     "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
   },
 
-  "LoggingSettings": {
+  "Logging": {
     "Path": "logs/",
     "Filename": "log_$start_time.log",
-    "ConsoleLogLevel": "TRACE",
-    "FileLogLevel": "TRACE"
+    "ConsoleLevel": "TRACE",
+    "Level": "TRACE"
   },
 
-  "EMailClientSettings": {
+  "EMailClient": {
     "Host": "mail.sh-edraft.de",
     "Port": "587",
     "UserName": "dev-srv@sh-edraft.de",
     "Credentials": "RmBOQX1eNFYiYjgsSid3fV1nelc2WA=="
   },
 
-  "DatabaseSettings": {
+  "Database": {
     "Host": "localhost",
     "User": "sh_cpl",
     "Password": "MHZhc0Y2bjhKc1VUMWV0Qw==",
@@ -31,7 +31,7 @@
     "AuthPlugin": "mysql_native_password"
   },
 
-  "TestSettings": {
+  "Test": {
     "Value": 20
   }
 }

example/custom/general/src/general/appsettings.json

@@ -1,15 +1,15 @@
 {
-  "TimeFormatSettings": {
+  "TimeFormat": {
     "DateFormat": "%Y-%m-%d",
     "TimeFormat": "%H:%M:%S",
     "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
     "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
   },
 
-  "LoggingSettings": {
+  "Logging": {
     "Path": "logs/",
     "Filename": "log_$start_time.log",
-    "ConsoleLogLevel": "ERROR",
-    "FileLogLevel": "WARN"
+    "ConsoleLevel": "ERROR",
+    "Level": "WARN"
   }
-}
+}

example/custom/general/src/general/general.json

@@ -1,5 +1,5 @@
 {
-  "ProjectSettings": {
+  "Project": {
     "Name": "general",
     "Version": {
       "Major": "2021",
@@ -30,7 +30,7 @@
     },
     "Classifiers": []
   },
-  "BuildSettings": {
+  "Build": {
     "ProjectType": "console",
     "SourcePath": "",
     "OutputPath": "dist",

example/custom/general/src/general/main.py

@@ -7,9 +7,9 @@ from test_startup_extension import TestStartupExtension
 
 def main():
     app_builder = ApplicationBuilder(Application)
-    app_builder.use_startup(Startup)
-    app_builder.use_extension(TestStartupExtension)
-    app_builder.use_extension(TestExtension)
+    app_builder.with_startup(Startup)
+    app_builder.with_extension(TestStartupExtension)
+    app_builder.with_extension(TestExtension)
     app_builder.build().run()
 
 

example/custom/general/src/general/startup.py

@@ -0,0 +1,23 @@
+from cpl import mail
+from cpl.application.abc import StartupABC
+from cpl.core.configuration import Configuration
+from cpl.core.environment import Environment
+from cpl.core.pipes import IPAddressPipe
+from cpl.dependency import ServiceCollection
+from test_service import TestService
+
+
+class Startup(StartupABC):
+
+    @staticmethod
+    def configure_configuration():
+        Configuration.add_json_file(f"appsettings.json")
+        Configuration.add_json_file(f"appsettings.{Environment.get_environment()}.json")
+        Configuration.add_json_file(f"appsettings.{Environment.get_host_name()}.json", optional=True)
+
+    @staticmethod
+    def configure_services(services: ServiceCollection):
+        services.add_logging()
+        services.add_module(mail)
+        services.add_transient(IPAddressPipe)
+        services.add_singleton(TestService)

example/custom/general/src/general/test_extension.py

@@ -0,0 +1,10 @@
+from cpl.application.abc import ApplicationExtensionABC
+from cpl.core.console import Console
+from cpl.dependency import ServiceProviderABC
+
+
+class TestExtension(ApplicationExtensionABC):
+
+    @staticmethod
+    def run(services: ServiceProviderABC):
+        Console.write_line("Hello World from App Extension")

example/custom/general/src/general/test_startup_extension.py

@@ -0,0 +1,14 @@
+from cpl.application.abc import StartupExtensionABC
+from cpl.core.console import Console
+from cpl.dependency import ServiceCollection
+
+
+class TestStartupExtension(StartupExtensionABC):
+
+    @staticmethod
+    def configure_configuration():
+        Console.write_line("config")
+
+    @staticmethod
+    def configure_services(services: ServiceCollection):
+        Console.write_line("services")

example/custom/query/main.py

@@ -0,0 +1,60 @@
+from cpl.core.console import Console
+from cpl.core.utils.benchmark import Benchmark
+from cpl.query.enumerable import Enumerable
+from cpl.query.immutable_list import ImmutableList
+from cpl.query.list import List
+from cpl.query.set import Set
+
+
+def _default():
+    Console.write_line(Enumerable.empty().to_list())
+
+    Console.write_line(Enumerable.range(0, 100).length)
+    Console.write_line(Enumerable.range(0, 100).to_list())
+
+    Console.write_line(Enumerable.range(0, 100).where(lambda x: x % 2 == 0).length)
+    Console.write_line(
+        Enumerable.range(0, 100).where(lambda x: x % 2 == 0).to_list().select(lambda x: str(x)).to_list()
+    )
+    Console.write_line(List)
+
+    s =Enumerable.range(0, 10).to_set()
+    Console.write_line(s)
+    s.add(1)
+    Console.write_line(s)
+
+    data = Enumerable(
+        [
+            {"name": "Alice", "age": 30},
+            {"name": "Dave", "age": 35},
+            {"name": "Charlie", "age": 25},
+            {"name": "Bob", "age": 25},
+        ]
+    )
+
+    Console.write_line(data.order_by(lambda x: x["age"]).to_list())
+    Console.write_line(data.order_by(lambda x: x["age"]).then_by(lambda x: x["name"]).to_list())
+    Console.write_line(data.order_by(lambda x: x["name"]).then_by(lambda x: x["age"]).to_list())
+
+
+def t_benchmark(data: list):
+    Benchmark.all("Enumerable", lambda: Enumerable(data).where(lambda x: x % 2 == 0).select(lambda x: x * 2).to_list())
+    Benchmark.all("Set", lambda: Set(data).where(lambda x: x % 2 == 0).select(lambda x: x * 2).to_list())
+    Benchmark.all("List", lambda: List(data).where(lambda x: x % 2 == 0).select(lambda x: x * 2).to_list())
+    Benchmark.all(
+        "ImmutableList", lambda: ImmutableList(data).where(lambda x: x % 2 == 0).select(lambda x: x * 2).to_list()
+    )
+    Benchmark.all("List comprehension", lambda: [x * 2 for x in data if x % 2 == 0])
+
+
+def main():
+    N = 10_000_000
+    data = list(range(N))
+    #t_benchmark(data)
+
+    Console.write_line()
+    _default()
+
+
+if __name__ == "__main__":
+    main()

example/custom/translation/cpl-workspace.json

@@ -1,5 +1,5 @@
 {
-  "WorkspaceSettings": {
+  "Workspace": {
     "DefaultProject": "translation",
     "Projects": {
       "translation": "src/translation/translation.json"

example/custom/translation/src/translation/application.py

@@ -18,8 +18,7 @@ class Application(ApplicationABC):
         self._translation.load_by_settings(config.get_configuration(TranslationSettings))
         self._translation.set_default_lang("de")
 
-    def configure(self):
-        pass
+    def configure(self): ...
 
     def main(self):
         Console.write_line(self._translate.transform("main.text.hello_world"))

example/custom/translation/src/translation/appsettings.json

@@ -1,16 +1,16 @@
 {
-  "TimeFormatSettings": {
+  "TimeFormat": {
     "DateFormat": "%Y-%m-%d",
     "TimeFormat": "%H:%M:%S",
     "DateTimeFormat": "%Y-%m-%d %H:%M:%S.%f",
     "DateTimeLogFormat": "%Y-%m-%d_%H-%M-%S"
   },
 
-  "LoggingSettings": {
+  "Logging": {
     "Path": "logs/",
     "Filename": "log_$start_time.log",
-    "ConsoleLogLevel": "ERROR",
-    "FileLogLevel": "WARN"
+    "ConsoleLevel": "ERROR",
+    "Level": "WARN"
   },
 
   "Translation": {

example/custom/translation/src/translation/main.py

@@ -6,7 +6,7 @@ from translation.startup import Startup
 
 def main():
     app_builder = ApplicationBuilder(Application)
-    app_builder.use_startup(Startup)
+    app_builder.with_startup(Startup)
     app_builder.build().run()
 
 

example/custom/translation/src/translation/startup.py

@@ -14,4 +14,4 @@ class Startup(StartupABC):
 
     def configure_services(self, services: ServiceCollection, environment: Environment) -> ServiceProviderABC:
         services.add_translation()
-        return services.build_service_provider()
+        return services.build()

example/custom/translation/src/translation/translation.json

@@ -1,5 +1,5 @@
 {
-  "ProjectSettings": {
+  "Project": {
     "Name": "translation",
     "Version": {
       "Major": "0",
@@ -25,7 +25,7 @@
     "PythonPath": {},
     "Classifiers": []
   },
-  "BuildSettings": {
+  "Build": {
     "ProjectType": "console",
     "SourcePath": "",
     "OutputPath": "../../dist",

install.sh

@@ -0,0 +1,61 @@
+#!/usr/bin/env bash
+set -euo pipefail
+
+# Find and combine requirements from src/cpl-*/requirements.txt,
+# filtering out lines whose *package name* starts with "cpl-".
+# Works with pinned versions, extras, markers, editable installs, and VCS refs.
+
+shopt -s nullglob
+
+req_files=(src/cpl-*/requirements.txt)
+if ((${#req_files[@]} == 0)); then
+  echo "No requirements files found at src/cpl-*/requirements.txt" >&2
+  exit 1
+fi
+
+tmp_combined="$(mktemp)"
+trap 'rm -f "$tmp_combined"' EXIT
+
+# Concatenate, trim comments/whitespace, filter out cpl-* packages, dedupe.
+# We keep non-package options/flags/constraints as-is.
+awk '
+  function trim(s){ sub(/^[[:space:]]+/,"",s); sub(/[[:space:]]+$/,"",s); return s }
+
+  {
+    line=$0
+    # drop full-line comments and strip inline comments
+    if (line ~ /^[[:space:]]*#/) next
+    sub(/#[^!].*$/,"",line)  # strip trailing comment (simple heuristic)
+    line=trim(line)
+    if (line == "") next
+
+    # Determine the package *name* even for "-e", extras, pins, markers, or VCS "@"
+    e = line
+    sub(/^-e[[:space:]]+/,"",e)           # remove editable prefix
+    # Tokenize up to the first of these separators: space, [ < > = ! ~ ; @
+    token = e
+    sub(/\[.*/,"",token)                  # remove extras quickly
+    n = split(token, a, /[<>=!~;@[:space:]]/)
+    name = tolower(a[1])
+
+    # If the first token (name) starts with "cpl-", skip this requirement
+    if (name ~ /^cpl-/) next
+
+    print line
+  }
+' "${req_files[@]}" | sort -u > "$tmp_combined"
+
+if ! [ -s "$tmp_combined" ]; then
+  echo "Nothing to install after filtering out cpl-* packages." >&2
+  exit 0
+fi
+
+echo "Installing dependencies (excluding cpl-*) from:"
+printf '  - %s\n' "${req_files[@]}"
+echo
+echo "Final set to install:"
+cat "$tmp_combined"
+echo
+
+# Use python -m pip for reliability; change to python3 if needed.
+python -m pip install -r "$tmp_combined"

src/cpl-api/cpl/api/__init__.py

@@ -0,0 +1,36 @@
+from cpl.dependency.service_collection import ServiceCollection as _ServiceCollection
+
+from .error import APIError, AlreadyExists, EndpointNotImplemented, Forbidden, NotFound, Unauthorized
+from .logger import APILogger
+from .settings import ApiSettings
+
+
+def add_api(collection: _ServiceCollection):
+    try:
+        from cpl.database import mysql
+
+        collection.add_module(mysql)
+    except ImportError as e:
+        from cpl.core.errors import dependency_error
+
+        dependency_error("cpl-database", e)
+
+    try:
+        from cpl import auth
+        from cpl.auth import permission
+
+        collection.add_module(auth)
+        collection.add_module(permission)
+    except ImportError as e:
+        from cpl.core.errors import dependency_error
+
+        dependency_error("cpl-auth", e)
+
+    from cpl.api.registry.policy import PolicyRegistry
+    from cpl.api.registry.route import RouteRegistry
+
+    collection.add_singleton(PolicyRegistry)
+    collection.add_singleton(RouteRegistry)
+
+
+_ServiceCollection.with_module(add_api, __name__)

src/cpl-api/cpl/api/abc/__init__.py

@@ -0,0 +1 @@
+from .asgi_middleware_abc import ASGIMiddleware

src/cpl-api/cpl/api/abc/asgi_middleware_abc.py

@@ -0,0 +1,15 @@
+from abc import ABC, abstractmethod
+
+from starlette.types import Scope, Receive, Send
+
+
+class ASGIMiddleware(ABC):
+    @abstractmethod
+    def __init__(self, app):
+        self._app = app
+
+    def _call_next(self, scope: Scope, receive: Receive, send: Send):
+        return self._app(scope, receive, send)
+
+    @abstractmethod
+    async def __call__(self, scope: Scope, receive: Receive, send: Send): ...

src/cpl-api/cpl/api/application/__init__.py

@@ -0,0 +1 @@
+from .web_app import WebApp

src/cpl-api/cpl/api/application/web_app.py

@@ -0,0 +1,249 @@
+import os
+from enum import Enum
+from typing import Mapping, Any, Callable, Self, Union
+
+import uvicorn
+from starlette.applications import Starlette
+from starlette.middleware import Middleware
+from starlette.middleware.cors import CORSMiddleware
+from starlette.requests import Request
+from starlette.responses import JSONResponse
+from starlette.types import ExceptionHandler
+
+from cpl import api, auth
+from cpl.api.error import APIError
+from cpl.api.logger import APILogger
+from cpl.api.middleware.authentication import AuthenticationMiddleware
+from cpl.api.middleware.authorization import AuthorizationMiddleware
+from cpl.api.middleware.logging import LoggingMiddleware
+from cpl.api.middleware.request import RequestMiddleware
+from cpl.api.model.api_route import ApiRoute
+from cpl.api.model.policy import Policy
+from cpl.api.model.validation_match import ValidationMatch
+from cpl.api.registry.policy import PolicyRegistry
+from cpl.api.registry.route import RouteRegistry
+from cpl.api.router import Router
+from cpl.api.settings import ApiSettings
+from cpl.api.typing import HTTPMethods, PartialMiddleware, PolicyResolver
+from cpl.application.abc.application_abc import ApplicationABC
+from cpl.core.configuration import Configuration
+from cpl.dependency.service_provider_abc import ServiceProviderABC
+
+
+PolicyInput = Union[dict[str, PolicyResolver], Policy]
+
+
+class WebApp(ApplicationABC):
+    def __init__(self, services: ServiceProviderABC):
+        super().__init__(services, [auth, api])
+        self._app: Starlette | None = None
+
+        self._logger = services.get_service(APILogger)
+
+        self._api_settings = Configuration.get(ApiSettings)
+        self._policies = services.get_service(PolicyRegistry)
+        self._routes = services.get_service(RouteRegistry)
+
+        self._middleware: list[Middleware] = [
+            Middleware(RequestMiddleware),
+            Middleware(LoggingMiddleware),
+        ]
+        self._exception_handlers: Mapping[Any, ExceptionHandler] = {
+            Exception: self._handle_exception,
+            APIError: self._handle_exception,
+        }
+
+    async def _handle_exception(self, request: Request, exc: Exception):
+        if isinstance(exc, APIError):
+            self._logger.error(exc)
+            return JSONResponse({"error": str(exc)}, status_code=exc.status_code)
+
+        if hasattr(request.state, "request_id"):
+            self._logger.error(f"Request {request.state.request_id}", exc)
+        else:
+            self._logger.error("Request unknown", exc)
+
+        return JSONResponse({"error": str(exc)}, status_code=500)
+
+    def _get_allowed_origins(self):
+        origins = self._api_settings.allowed_origins
+
+        if origins is None or origins == "":
+            self._logger.warning("No allowed origins specified, allowing all origins")
+            return ["*"]
+
+        self._logger.debug(f"Allowed origins: {origins}")
+        return origins.split(",")
+
+    def with_database(self) -> Self:
+        self.with_migrations()
+        self.with_seeders()
+        return self
+
+    def with_app(self, app: Starlette) -> Self:
+        assert app is not None, "app must not be None"
+        assert isinstance(app, Starlette), "app must be an instance of Starlette"
+        self._app = app
+        return self
+
+    def _check_for_app(self):
+        if self._app is not None:
+            raise ValueError("App is already set, cannot add routes or middleware")
+
+    def with_routes_directory(self, directory: str) -> Self:
+        self._check_for_app()
+        assert directory is not None, "directory must not be None"
+
+        base = directory.replace("/", ".").replace("\\", ".")
+
+        for filename in os.listdir(directory):
+            if not filename.endswith(".py") or filename == "__init__.py":
+                continue
+
+            __import__(f"{base}.{filename[:-3]}")
+
+        return self
+
+    def with_routes(
+        self,
+        routes: list[ApiRoute],
+        method: HTTPMethods,
+        authentication: bool = False,
+        roles: list[str | Enum] = None,
+        permissions: list[str | Enum] = None,
+        policies: list[str] = None,
+        match: ValidationMatch = None,
+    ) -> Self:
+        self._check_for_app()
+        assert self._routes is not None, "routes must not be None"
+        assert all(isinstance(route, ApiRoute) for route in routes), "all routes must be of type ApiRoute"
+        for route in routes:
+            self.with_route(
+                route.path,
+                route.fn,
+                method,
+                authentication,
+                roles,
+                permissions,
+                policies,
+                match,
+            )
+        return self
+
+    def with_route(
+        self,
+        path: str,
+        fn: Callable[[Request], Any],
+        method: HTTPMethods,
+        authentication: bool = False,
+        roles: list[str | Enum] = None,
+        permissions: list[str | Enum] = None,
+        policies: list[str] = None,
+        match: ValidationMatch = None,
+    ) -> Self:
+        self._check_for_app()
+        assert path is not None, "path must not be None"
+        assert fn is not None, "fn must not be None"
+        assert method in [
+            "GET",
+            "HEAD",
+            "POST",
+            "PUT",
+            "PATCH",
+            "DELETE",
+            "OPTIONS",
+        ], "method must be a valid HTTP method"
+
+        Router.route(path, method, registry=self._routes)(fn)
+
+        if authentication:
+            Router.authenticate()(fn)
+
+        if roles or permissions or policies:
+            Router.authorize(roles, permissions, policies, match)(fn)
+
+        return self
+
+    def with_middleware(self, middleware: PartialMiddleware) -> Self:
+        self._check_for_app()
+
+        if isinstance(middleware, Middleware):
+            self._middleware.append(middleware)
+        elif callable(middleware):
+            self._middleware.append(Middleware(middleware))
+        else:
+            raise ValueError("middleware must be of type starlette.middleware.Middleware or a callable")
+
+        return self
+
+    def with_authentication(self) -> Self:
+        self.with_middleware(AuthenticationMiddleware)
+        return self
+
+    def with_authorization(self, *policies: list[PolicyInput] | PolicyInput) -> Self:
+        if policies:
+            _policies = []
+
+            if not isinstance(policies, list):
+                policies = list(policies)
+
+            for i, policy in enumerate(policies):
+                if isinstance(policy, dict):
+                    for name, resolver in policy.items():
+                        if not isinstance(name, str):
+                            self._logger.warning(f"Skipping policy at index {i}, name must be a string")
+                            continue
+
+                        if not callable(resolver):
+                            self._logger.warning(f"Skipping policy {name}, resolver must be callable")
+                            continue
+
+                        _policies.append(Policy(name, resolver))
+                    continue
+
+                _policies.append(policy)
+
+            self._policies.extend(_policies)
+
+        self.with_middleware(AuthorizationMiddleware)
+        return self
+
+    def _validate_policies(self):
+        for rule in Router.get_authorization_rules():
+            for policy_name in rule["policies"]:
+                policy = self._policies.get(policy_name)
+                if not policy:
+                    self._logger.fatal(f"Authorization policy '{policy_name}' not found")
+
+    async def main(self):
+        self._logger.debug(f"Preparing API")
+        self._validate_policies()
+
+        if self._app is None:
+            routes = [route.to_starlette(self._services.inject) for route in self._routes.all()]
+
+            app = Starlette(
+                routes=routes,
+                middleware=[
+                    *self._middleware,
+                    Middleware(
+                        CORSMiddleware,
+                        allow_origins=self._get_allowed_origins(),
+                        allow_methods=["*"],
+                        allow_headers=["*"],
+                    ),
+                ],
+                exception_handlers=self._exception_handlers,
+            )
+        else:
+            app = self._app
+
+        self._logger.info(f"Start API on {self._api_settings.host}:{self._api_settings.port}")
+
+        config = uvicorn.Config(
+            app, host=self._api_settings.host, port=self._api_settings.port, log_config=None, loop="asyncio"
+        )
+        server = uvicorn.Server(config)
+        await server.serve()
+
+        self._logger.info("Shutdown API")

src/cpl-api/cpl/api/error.py

@@ -0,0 +1,46 @@
+from http.client import HTTPException
+
+from starlette.responses import JSONResponse
+from starlette.types import Scope, Receive, Send
+
+
+class APIError(HTTPException):
+    status_code = 500
+
+    def __init__(self, message: str = ""):
+        super().__init__(self.status_code, message)
+        self._message = message
+
+    @property
+    def error_message(self) -> str:
+        if self._message:
+            return f"{type(self).__name__}: {self._message}"
+
+        return f"{type(self).__name__}"
+
+    async def asgi_response(self, scope: Scope, receive: Receive, send: Send):
+        r = JSONResponse({"error": self.error_message}, status_code=self.status_code)
+        return await r(scope, receive, send)
+
+    def response(self):
+        return JSONResponse({"error": self.error_message}, status_code=self.status_code)
+
+
+class Unauthorized(APIError):
+    status_code = 401
+
+
+class Forbidden(APIError):
+    status_code = 403
+
+
+class NotFound(APIError):
+    status_code = 404
+
+
+class AlreadyExists(APIError):
+    status_code = 409
+
+
+class EndpointNotImplemented(APIError):
+    status_code = 501

src/cpl-api/cpl/api/logger.py

@@ -0,0 +1,7 @@
+from cpl.core.log.wrapped_logger import WrappedLogger
+
+
+class APILogger(WrappedLogger):
+
+    def __init__(self):
+        WrappedLogger.__init__(self, "api")

src/cpl-api/cpl/api/middleware/__init__.py

@@ -0,0 +1,4 @@
+from .authentication import AuthenticationMiddleware
+from .authorization import AuthorizationMiddleware
+from .logging import LoggingMiddleware
+from .request import RequestMiddleware

src/cpl-api/cpl/api/middleware/authentication.py

@@ -0,0 +1,80 @@
+from keycloak import KeycloakAuthenticationError
+from starlette.types import Scope, Receive, Send
+
+from cpl.api.abc.asgi_middleware_abc import ASGIMiddleware
+from cpl.api.error import Unauthorized
+from cpl.api.logger import APILogger
+from cpl.api.middleware.request import get_request
+from cpl.api.router import Router
+from cpl.auth.keycloak import KeycloakClient
+from cpl.auth.schema import AuthUserDao, AuthUser
+from cpl.core.ctx import set_user
+from cpl.dependency import ServiceProviderABC
+
+
+class AuthenticationMiddleware(ASGIMiddleware):
+
+    @ServiceProviderABC.inject
+    def __init__(self, app, logger: APILogger, keycloak: KeycloakClient, user_dao: AuthUserDao):
+        ASGIMiddleware.__init__(self, app)
+
+        self._logger = logger
+
+        self._keycloak = keycloak
+        self._user_dao = user_dao
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send):
+        request = get_request()
+        url = request.url.path
+
+        if url not in Router.get_auth_required_routes():
+            self._logger.trace(f"No authentication required for {url}")
+            return await self._app(scope, receive, send)
+
+        if not request.headers.get("Authorization"):
+            self._logger.debug(f"Unauthorized access to {url}, missing Authorization header")
+            return await Unauthorized(f"Missing header Authorization").asgi_response(scope, receive, send)
+
+        auth_header = request.headers.get("Authorization", None)
+        if not auth_header or not auth_header.startswith("Bearer "):
+            return await Unauthorized("Invalid Authorization header").asgi_response(scope, receive, send)
+
+        token = auth_header.split("Bearer ")[1]
+        if not await self._verify_login(token):
+            self._logger.debug(f"Unauthorized access to {url}, invalid token")
+            return await Unauthorized("Invalid token").asgi_response(scope, receive, send)
+
+        # check user exists in db, if not create
+        keycloak_id = self._keycloak.get_user_id(token)
+        if keycloak_id is None:
+            return await Unauthorized("Failed to get user id from token").asgi_response(scope, receive, send)
+
+        user = await self._get_or_crate_user(keycloak_id)
+        if user.deleted:
+            self._logger.debug(f"Unauthorized access to {url}, user is deleted")
+            return await Unauthorized("User is deleted").asgi_response(scope, receive, send)
+
+        request.state.user = user
+        set_user(user)
+
+        return await self._call_next(scope, receive, send)
+
+    async def _get_or_crate_user(self, keycloak_id: str) -> AuthUser:
+        existing = await self._user_dao.find_by_keycloak_id(keycloak_id)
+        if existing is not None:
+            return existing
+
+        user = AuthUser(0, keycloak_id)
+        uid = await self._user_dao.create(user)
+        return await self._user_dao.get_by_id(uid)
+
+    async def _verify_login(self, token: str) -> bool:
+        try:
+            token_info = self._keycloak.introspect(token)
+            return token_info.get("active", False)
+        except KeycloakAuthenticationError as e:
+            self._logger.debug(f"Keycloak authentication error: {e}")
+            return False
+        except Exception as e:
+            self._logger.error(f"Unexpected error during token verification: {e}")
+            return False

src/cpl-api/cpl/api/middleware/authorization.py

@@ -0,0 +1,73 @@
+from starlette.types import Scope, Receive, Send
+
+from cpl.api.abc.asgi_middleware_abc import ASGIMiddleware
+from cpl.api.error import Unauthorized, Forbidden
+from cpl.api.logger import APILogger
+from cpl.api.middleware.request import get_request
+from cpl.api.model.validation_match import ValidationMatch
+from cpl.api.registry.policy import PolicyRegistry
+from cpl.api.router import Router
+from cpl.auth.schema._administration.auth_user_dao import AuthUserDao
+from cpl.core.ctx.user_context import get_user
+from cpl.dependency.service_provider_abc import ServiceProviderABC
+
+
+class AuthorizationMiddleware(ASGIMiddleware):
+
+    @ServiceProviderABC.inject
+    def __init__(self, app, logger: APILogger, policies: PolicyRegistry, user_dao: AuthUserDao):
+        ASGIMiddleware.__init__(self, app)
+
+        self._logger = logger
+
+        self._policies = policies
+        self._user_dao = user_dao
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send):
+        request = get_request()
+        url = request.url.path
+
+        if url not in Router.get_authorization_rules_paths():
+            self._logger.trace(f"No authorization required for {url}")
+            return await self._app(scope, receive, send)
+
+        user = get_user()
+        if not user:
+            return await Unauthorized(f"Unknown user").asgi_response(scope, receive, send)
+
+        roles = await user.roles
+        request.state.roles = roles
+        role_names = [r.name for r in roles]
+
+        perms = await user.permissions
+        request.state.permissions = perms
+        perm_names = [p.name for p in perms]
+
+        for rule in Router.get_authorization_rules():
+            match = rule["match"]
+            if rule["roles"]:
+                if match == ValidationMatch.all and not all(r in role_names for r in rule["roles"]):
+                    return await Forbidden(f"missing roles: {rule["roles"]}").asgi_response(scope, receive, send)
+                if match == ValidationMatch.any and not any(r in role_names for r in rule["roles"]):
+                    return await Forbidden(f"missing roles: {rule["roles"]}").asgi_response(scope, receive, send)
+
+            if rule["permissions"]:
+                if match == ValidationMatch.all and not all(p in perm_names for p in rule["permissions"]):
+                    return await Forbidden(f"missing permissions: {rule["permissions"]}").asgi_response(
+                        scope, receive, send
+                    )
+                if match == ValidationMatch.any and not any(p in perm_names for p in rule["permissions"]):
+                    return await Forbidden(f"missing permissions: {rule["permissions"]}").asgi_response(
+                        scope, receive, send
+                    )
+
+            for policy_name in rule["policies"]:
+                policy = self._policies.get(policy_name)
+                if not policy:
+                    self._logger.warning(f"Authorization policy '{policy_name}' not found")
+                    continue
+
+                if not await policy.resolve(user):
+                    return await Forbidden(f"policy {policy.name} failed").asgi_response(scope, receive, send)
+
+        return await self._call_next(scope, receive, send)

src/cpl-api/cpl/api/middleware/logging.py

@@ -0,0 +1,87 @@
+import time
+
+from starlette.requests import Request
+from starlette.types import Receive, Scope, Send
+
+from cpl.api.abc.asgi_middleware_abc import ASGIMiddleware
+from cpl.api.logger import APILogger
+from cpl.api.middleware.request import get_request
+from cpl.dependency import ServiceProviderABC
+
+
+class LoggingMiddleware(ASGIMiddleware):
+
+    @ServiceProviderABC.inject
+    def __init__(self, app, logger: APILogger):
+        ASGIMiddleware.__init__(self, app)
+
+        self._logger = logger
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send):
+        if scope["type"] != "http":
+            await self._call_next(scope, receive, send)
+            return
+
+        request = get_request()
+        await self._log_request(request)
+        start_time = time.time()
+
+        response_body = b""
+        status_code = 500
+
+        async def send_wrapper(message):
+            nonlocal response_body, status_code
+            if message["type"] == "http.response.start":
+                status_code = message["status"]
+            if message["type"] == "http.response.body":
+                response_body += message.get("body", b"")
+            await send(message)
+
+        await self._call_next(scope, receive, send_wrapper)
+
+        duration = (time.time() - start_time) * 1000
+        await self._log_after_request(request, status_code, duration)
+
+    @staticmethod
+    def _filter_relevant_headers(headers: dict) -> dict:
+        relevant_keys = {
+            "content-type",
+            "host",
+            "connection",
+            "user-agent",
+            "origin",
+            "referer",
+            "accept",
+        }
+        return {key: value for key, value in headers.items() if key in relevant_keys}
+
+    async def _log_request(self, request: Request):
+        self._logger.debug(
+            f"Request {getattr(request.state, 'request_id', '-')}: {request.method}@{request.url.path} from {request.client.host}"
+        )
+
+        from cpl.core.ctx.user_context import get_user
+
+        user = get_user()
+
+        request_info = {
+            "headers": self._filter_relevant_headers(dict(request.headers)),
+            "args": dict(request.query_params),
+            "form-data": (
+                await request.form()
+                if request.headers.get("content-type") == "application/x-www-form-urlencoded"
+                else None
+            ),
+            "payload": (await request.json() if request.headers.get("content-length") == "0" else None),
+            "user": f"{user.id}-{user.keycloak_id}" if user else None,
+            "files": (
+                {key: file.filename for key, file in (await request.form()).items()} if await request.form() else None
+            ),
+        }
+
+        self._logger.trace(f"Request {getattr(request.state, 'request_id', '-')}: {request_info}")
+
+    async def _log_after_request(self, request: Request, status_code: int, duration: float):
+        self._logger.info(
+            f"Request finished {getattr(request.state, 'request_id', '-')}: {status_code}-{request.method}@{request.url.path} from {request.client.host} in {duration:.2f}ms"
+        )

src/cpl-api/cpl/api/middleware/request.py

@@ -0,0 +1,56 @@
+import time
+from contextvars import ContextVar
+from typing import Optional, Union
+from uuid import uuid4
+
+from starlette.requests import Request
+from starlette.types import Scope, Receive, Send
+
+from cpl.api.abc.asgi_middleware_abc import ASGIMiddleware
+from cpl.api.logger import APILogger
+from cpl.api.typing import TRequest
+from cpl.dependency import ServiceProviderABC
+
+_request_context: ContextVar[Union[TRequest, None]] = ContextVar("request", default=None)
+
+
+class RequestMiddleware(ASGIMiddleware):
+
+    @ServiceProviderABC.inject
+    def __init__(self, app, logger: APILogger):
+        ASGIMiddleware.__init__(self, app)
+
+        self._logger = logger
+
+        self._ctx_token = None
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send):
+        request = Request(scope, receive, send)
+        await self.set_request_data(request)
+
+        try:
+            await self._app(scope, receive, send)
+        finally:
+            await self.clean_request_data()
+
+    async def set_request_data(self, request: TRequest):
+        request.state.request_id = uuid4()
+        request.state.start_time = time.time()
+        self._logger.trace(f"Set new current request: {request.state.request_id}")
+
+        self._ctx_token = _request_context.set(request)
+
+    async def clean_request_data(self):
+        request = get_request()
+        if request is None:
+            return
+
+        if self._ctx_token is None:
+            return
+
+        self._logger.trace(f"Clearing current request: {request.state.request_id}")
+        _request_context.reset(self._ctx_token)
+
+
+def get_request() -> Optional[TRequest]:
+    return _request_context.get()

src/cpl-api/cpl/api/model/__init__.py

@@ -0,0 +1,3 @@
+from .api_route import ApiRoute
+from .policy import Policy
+from .validation_match import ValidationMatch

src/cpl-api/cpl/api/model/api_route.py

@@ -0,0 +1,43 @@
+from typing import Callable
+
+from starlette.routing import Route
+
+from cpl.api.typing import HTTPMethods
+
+
+class ApiRoute:
+
+    def __init__(self, path: str, fn: Callable, method: HTTPMethods, **kwargs):
+        self._path = path
+        self._fn = fn
+        self._method = method
+
+        self._kwargs = kwargs
+
+    @property
+    def name(self) -> str:
+        return self._fn.__name__
+
+    @property
+    def fn(self) -> Callable:
+        return self._fn
+
+    @property
+    def path(self) -> str:
+        return self._path
+
+    @property
+    def method(self) -> HTTPMethods:
+        return self._method
+
+    @property
+    def kwargs(self) -> dict:
+        return self._kwargs
+
+    def to_starlette(self, wrap_endpoint: Callable = None) -> Route:
+        return Route(
+            self._path,
+            self._fn if not wrap_endpoint else wrap_endpoint(self._fn),
+            methods=[self._method],
+            **self._kwargs,
+        )

src/cpl-api/cpl/api/model/policy.py

@@ -0,0 +1,34 @@
+from asyncio import iscoroutinefunction
+from typing import Optional
+
+from cpl.api.typing import PolicyResolver
+from cpl.core.ctx import get_user
+
+
+class Policy:
+    def __init__(
+        self,
+        name: str,
+        resolver: PolicyResolver = None,
+    ):
+        self._name = name
+        self._resolver: Optional[PolicyResolver] = resolver
+
+    @property
+    def name(self) -> str:
+        return self._name
+
+    @property
+    def resolvers(self) -> PolicyResolver:
+        return self._resolver
+
+    async def resolve(self, *args, **kwargs) -> bool:
+        if not self._resolver:
+            return True
+
+        if callable(self._resolver):
+            if iscoroutinefunction(self._resolver):
+                return await self._resolver(get_user())
+
+            return self._resolver(get_user())
+        return False

src/cpl-api/cpl/api/model/validation_match.py

@@ -0,0 +1,6 @@
+from enum import Enum
+
+
+class ValidationMatch(Enum):
+    any = "any"
+    all = "all"

src/cpl-api/cpl/api/registry/__init__.py

@@ -0,0 +1,2 @@
+from .policy import PolicyRegistry
+from .route import RouteRegistry

src/cpl-api/cpl/api/registry/policy.py

@@ -0,0 +1,28 @@
+from typing import Optional
+
+from cpl.api.model.policy import Policy
+from cpl.core.abc.registry_abc import RegistryABC
+
+
+class PolicyRegistry(RegistryABC):
+
+    def __init__(self):
+        RegistryABC.__init__(self)
+
+    def extend(self, items: list[Policy]):
+        for policy in items:
+            self.add(policy)
+
+    def add(self, item: Policy):
+        assert isinstance(item, Policy), "policy must be an instance of Policy"
+
+        if item.name in self._items:
+            raise ValueError(f"Policy {item.name} is already registered")
+
+        self._items[item.name] = item
+
+    def get(self, key: str) -> Optional[Policy]:
+        return self._items.get(key)
+
+    def all(self) -> list[Policy]:
+        return list(self._items.values())

src/cpl-api/cpl/api/registry/route.py

@@ -0,0 +1,32 @@
+from typing import Optional
+
+from cpl.api.model.api_route import ApiRoute
+from cpl.core.abc.registry_abc import RegistryABC
+
+
+class RouteRegistry(RegistryABC):
+
+    def __init__(self):
+        RegistryABC.__init__(self)
+
+    def extend(self, items: list[ApiRoute]):
+        for policy in items:
+            self.add(policy)
+
+    def add(self, item: ApiRoute):
+        assert isinstance(item, ApiRoute), "route must be an instance of ApiRoute"
+
+        if item.path in self._items:
+            raise ValueError(f"ApiRoute {item.path} is already registered")
+
+        self._items[item.path] = item
+
+    def set(self, item: ApiRoute):
+        assert isinstance(item, ApiRoute), "route must be an instance of ApiRoute"
+        self._items[item.path] = item
+
+    def get(self, key: str) -> Optional[ApiRoute]:
+        return self._items.get(key)
+
+    def all(self) -> list[ApiRoute]:
+        return list(self._items.values())

src/cpl-api/cpl/api/router.py

@@ -0,0 +1,164 @@
+from enum import Enum
+
+from cpl.api.model.validation_match import ValidationMatch
+from cpl.api.registry.route import RouteRegistry
+from cpl.api.typing import HTTPMethods
+
+
+class Router:
+    _auth_required: list[str] = []
+    _authorization_rules: dict[str, dict] = {}
+
+    @classmethod
+    def get_auth_required_routes(cls) -> list[str]:
+        return cls._auth_required
+
+    @classmethod
+    def get_authorization_rules_paths(cls) -> list[str]:
+        return list(cls._authorization_rules.keys())
+
+    @classmethod
+    def get_authorization_rules(cls) -> list[dict]:
+        return list(cls._authorization_rules.values())
+
+    @classmethod
+    def authenticate(cls):
+        """
+        Decorator to mark a route as requiring authentication.
+        Usage:
+            @Route.authenticate()
+            @Route.get("/example")
+            async def example_endpoint(request: TRequest):
+                ...
+        """
+
+        def inner(fn):
+            route_path = getattr(fn, "_route_path", None)
+            if route_path and route_path not in cls._auth_required:
+                cls._auth_required.append(route_path)
+            return fn
+
+        return inner
+
+    @classmethod
+    def authorize(
+        cls,
+        roles: list[str | Enum] = None,
+        permissions: list[str | Enum] = None,
+        policies: list[str] = None,
+        match: ValidationMatch = None,
+    ):
+        """
+        Decorator to mark a route as requiring authorization.
+        Usage:
+            @Route.authorize()
+            @Route.get("/example")
+            async def example_endpoint(request: TRequest):
+                ...
+        """
+        assert roles is None or isinstance(roles, list), "roles must be a list of strings"
+        assert permissions is None or isinstance(permissions, list), "permissions must be a list of strings"
+        assert policies is None or isinstance(policies, list), "policies must be a list of strings"
+        assert match is None or isinstance(match, ValidationMatch), "match must be an instance of ValidationMatch"
+
+        if roles is not None:
+            for role in roles:
+                if isinstance(role, Enum):
+                    roles[roles.index(role)] = role.value
+
+        if permissions is not None:
+            for perm in permissions:
+                if isinstance(perm, Enum):
+                    permissions[permissions.index(perm)] = perm.value
+
+        def inner(fn):
+            path = getattr(fn, "_route_path", None)
+            if not path:
+                return fn
+
+            if path in cls._authorization_rules:
+                raise ValueError(f"Route {path} is already registered for authorization")
+
+            cls._authorization_rules[path] = {
+                "roles": roles or [],
+                "permissions": permissions or [],
+                "policies": policies or [],
+                "match": match or ValidationMatch.all,
+            }
+
+            return fn
+
+        return inner
+
+    @classmethod
+    def route(cls, path: str, method: HTTPMethods, registry: RouteRegistry = None, **kwargs):
+        from cpl.api.model.api_route import ApiRoute
+
+        if not registry:
+            from cpl.dependency.service_provider_abc import ServiceProviderABC
+
+            routes = ServiceProviderABC.get_global_service(RouteRegistry)
+        else:
+            routes = registry
+
+        def inner(fn):
+            routes.add(ApiRoute(path, fn, method, **kwargs))
+            setattr(fn, "_route_path", path)
+            return fn
+
+        return inner
+
+    @classmethod
+    def get(cls, path: str, **kwargs):
+        return cls.route(path, "GET", **kwargs)
+
+    @classmethod
+    def head(cls, path: str, **kwargs):
+        return cls.route(path, "HEAD", **kwargs)
+
+    @classmethod
+    def post(cls, path: str, **kwargs):
+        return cls.route(path, "POST", **kwargs)
+
+    @classmethod
+    def put(cls, path: str, **kwargs):
+        return cls.route(path, "PUT", **kwargs)
+
+    @classmethod
+    def patch(cls, path: str, **kwargs):
+        return cls.route(path, "PATCH", **kwargs)
+
+    @classmethod
+    def delete(cls, path: str, **kwargs):
+        return cls.route(path, "DELETE", **kwargs)
+
+    @classmethod
+    def override(cls):
+        """
+        Decorator to override an existing route with the same path.
+        Usage:
+            @Route.override()
+            @Route.get("/example")
+            async def example_endpoint(request: TRequest):
+                ...
+        """
+
+        from cpl.api.model.api_route import ApiRoute
+        from cpl.dependency.service_provider_abc import ServiceProviderABC
+
+        routes = ServiceProviderABC.get_global_service(RouteRegistry)
+
+        def inner(fn):
+            path = getattr(fn, "_route_path", None)
+            if path is None:
+                raise ValueError("Cannot override a route that has not been registered yet")
+
+            route = routes.get(path)
+            if route is None:
+                raise ValueError(f"Cannot override a route that does not exist: {path}")
+
+            routes.add(ApiRoute(path, fn, route.method, **route.kwargs))
+            setattr(fn, "_route_path", path)
+            return fn
+
+        return inner