Compare commits

..

14 Commits

Author SHA1 Message Date
4f698269b5 Fixed api build
All checks were successful
Build on push / prepare (push) Successful in 9s
Build on push / core (push) Successful in 18s
Build on push / query (push) Successful in 17s
Build on push / dependency (push) Successful in 18s
Build on push / application (push) Successful in 15s
Build on push / database (push) Successful in 18s
Build on push / translation (push) Successful in 18s
Build on push / mail (push) Successful in 19s
Build on push / auth (push) Successful in 15s
Build on push / api (push) Successful in 14s
2025-09-19 21:12:33 +02:00
ddc62dfb9a Added api & route handling
Some checks failed
Build on push / prepare (push) Successful in 9s
Build on push / core (push) Successful in 19s
Build on push / query (push) Successful in 19s
Build on push / dependency (push) Successful in 17s
Build on push / application (push) Successful in 15s
Build on push / database (push) Successful in 18s
Build on push / mail (push) Successful in 19s
Build on push / translation (push) Successful in 23s
Build on push / auth (push) Successful in 16s
Build on push / api (push) Failing after 14s
2025-09-19 21:03:33 +02:00
1a67318091 Config model options handling. Closes #185
All checks were successful
Build on push / prepare (push) Successful in 10s
Build on push / core (push) Successful in 19s
Build on push / query (push) Successful in 19s
Build on push / dependency (push) Successful in 25s
Build on push / translation (push) Successful in 17s
Build on push / database (push) Successful in 20s
Build on push / application (push) Successful in 21s
Build on push / mail (push) Successful in 20s
Build on push / auth (push) Successful in 14s
2025-09-19 17:47:49 +02:00
2be58f6577 Introduced fernet to credential manager. Closes #183
All checks were successful
Build on push / prepare (push) Successful in 10s
Build on push / core (push) Successful in 19s
Build on push / query (push) Successful in 22s
Build on push / dependency (push) Successful in 15s
Build on push / application (push) Successful in 20s
Build on push / database (push) Successful in 21s
Build on push / translation (push) Successful in 21s
Build on push / mail (push) Successful in 22s
Build on push / auth (push) Successful in 18s
2025-09-19 15:01:16 +02:00
9c6078f4fd with_logging & logger level fix
All checks were successful
Build on push / prepare (push) Successful in 8s
Build on push / core (push) Successful in 18s
Build on push / query (push) Successful in 18s
Build on push / dependency (push) Successful in 14s
Build on push / translation (push) Successful in 15s
Build on push / database (push) Successful in 17s
Build on push / application (push) Successful in 18s
Build on push / mail (push) Successful in 20s
Build on push / auth (push) Successful in 18s
2025-09-17 22:18:38 +02:00
dfdc31512d App with extension functions
All checks were successful
Build on push / prepare (push) Successful in 9s
Build on push / core (push) Successful in 17s
Build on push / query (push) Successful in 17s
Build on push / dependency (push) Successful in 14s
Build on push / translation (push) Successful in 14s
Build on push / database (push) Successful in 18s
Build on push / mail (push) Successful in 19s
Build on push / application (push) Successful in 22s
Build on push / auth (push) Successful in 14s
2025-09-17 21:56:47 +02:00
ab7ff7da93 Made startup/app extensions static 2025-09-17 20:54:21 +02:00
41087a838b Removed pass from empty functions
All checks were successful
Build on push / prepare (push) Successful in 9s
Build on push / core (push) Successful in 17s
Build on push / query (push) Successful in 17s
Build on push / dependency (push) Successful in 17s
Build on push / translation (push) Successful in 14s
Build on push / mail (push) Successful in 18s
Build on push / database (push) Successful in 18s
Build on push / application (push) Successful in 24s
Build on push / auth (push) Successful in 14s
2025-09-17 20:49:15 +02:00
836b92ccbf Further console test 2025-09-17 20:44:25 +02:00
8aaba22940 Improved application structure
All checks were successful
Build on push / prepare (push) Successful in 11s
Build on push / core (push) Successful in 22s
Build on push / query (push) Successful in 22s
Build on push / dependency (push) Successful in 20s
Build on push / database (push) Successful in 20s
Build on push / translation (push) Successful in 20s
Build on push / application (push) Successful in 22s
Build on push / mail (push) Successful in 23s
Build on push / auth (push) Successful in 16s
2025-09-17 19:23:14 +02:00
504dc5e188 Added auth & improved database
All checks were successful
Build on push / prepare (push) Successful in 9s
Build on push / query (push) Successful in 18s
Build on push / core (push) Successful in 24s
Build on push / dependency (push) Successful in 17s
Build on push / database (push) Successful in 15s
Build on push / translation (push) Successful in 15s
Build on push / mail (push) Successful in 18s
Build on push / application (push) Successful in 19s
Build on push / auth (push) Successful in 16s
2025-09-17 12:21:32 +02:00
4625b626e6 Added dao base
All checks were successful
Build on push / prepare (push) Successful in 8s
Build on push / query (push) Successful in 16s
Build on push / core (push) Successful in 23s
Build on push / translation (push) Successful in 14s
Build on push / mail (push) Successful in 14s
2025-09-17 00:12:25 +02:00
58dbd3ed1e Cleanup for mysql
All checks were successful
Build on push / prepare (push) Successful in 8s
Build on push / core (push) Successful in 19s
Build on push / query (push) Successful in 18s
Build on push / translation (push) Successful in 15s
Build on push / mail (push) Successful in 17s
2025-09-16 20:21:33 +02:00
cd7dfaf2b4 Fixed spinner thread & remove log thread
All checks were successful
Build on push / prepare (push) Successful in 8s
Build on push / core (push) Successful in 18s
Build on push / query (push) Successful in 19s
Build on push / translation (push) Successful in 18s
Build on push / mail (push) Successful in 18s
2025-09-16 18:41:57 +02:00
227 changed files with 5307 additions and 1809 deletions

View File

@@ -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

1
.gitignore vendored
View File

@@ -113,6 +113,7 @@ venv.bak/
# Custom Environments
cpl-env/
.secret
# Spyder project settings
.spyderproject

153
README.md
View File

@@ -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

View File

@@ -0,0 +1,7 @@
from cpl.core.log.logger import Logger
class APILogger(Logger):
def __init__(self, source: str):
Logger.__init__(self, source, "api")

View File

@@ -0,0 +1,13 @@
from typing import Optional
from cpl.core.configuration import ConfigurationModelABC
class ApiSettings(ConfigurationModelABC):
def __init__(self, src: Optional[dict] = None):
super().__init__(src)
self.option("host", str, "0.0.0.0")
self.option("port", int, 5000)
self.option("allowed_origins", list[str])

View File

@@ -0,0 +1,25 @@
from http.client import HTTPException
class APIError(HTTPException):
status_code = 500
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

View File

@@ -0,0 +1,65 @@
import time
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.requests import Request
from starlette.responses import Response
from cpl.api.api_logger import APILogger
_logger = APILogger(__name__)
class LoggingMiddleware(BaseHTTPMiddleware):
async def dispatch(self, request: Request, call_next):
await self._log_request(request)
response = await call_next(request)
await self._log_after_request(request, response)
return response
@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}
@classmethod
async def _log_request(cls, request: Request):
_logger.debug(
f"Request {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": cls._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
),
}
_logger.trace(f"Request {request.state.request_id}: {request_info}")
@staticmethod
async def _log_after_request(request: Request, response: Response):
duration = (time.time() - request.state.start_time) * 1000
_logger.info(
f"Request finished {request.state.request_id}: {response.status_code}-{request.method}@{request.url.path} from {request.client.host} in {duration:.2f}ms"
)

View File

@@ -0,0 +1,48 @@
import time
from contextvars import ContextVar
from typing import Optional, Union
from uuid import uuid4
from starlette.middleware.base import BaseHTTPMiddleware
from starlette.websockets import WebSocket
from cpl.api.api_logger import APILogger
from cpl.api.typing import TRequest
_request_context: ContextVar[Union[TRequest, None]] = ContextVar("request", default=None)
_logger = APILogger(__name__)
class RequestMiddleware(BaseHTTPMiddleware):
_request_token = {}
_user_token = {}
@classmethod
async def set_request_data(cls, request: TRequest):
request.state.request_id = uuid4()
request.state.start_time = time.time()
_logger.trace(f"Set new current request: {request.state.request_id}")
cls._request_token[request.state.request_id] = _request_context.set(request)
@classmethod
async def clean_request_data(cls):
request = get_request()
if request is None:
return
if request.state.request_id in cls._request_token:
_request_context.reset(cls._request_token[request.state.request_id])
async def dispatch(self, request: TRequest, call_next):
await self.set_request_data(request)
try:
response = await call_next(request)
return response
finally:
await self.clean_request_data()
def get_request() -> Optional[Union[TRequest, WebSocket]]:
return _request_context.get()

View File

@@ -0,0 +1,60 @@
from starlette.routing import Route
class Router:
_registered_routes: list[Route] = []
@classmethod
def get_routes(cls) -> list[Route]:
return cls._registered_routes
@classmethod
def route(cls, path=None, **kwargs):
def inner(fn):
cls._registered_routes.append(Route(path, fn, **kwargs))
setattr(fn, "_route_path", path)
return fn
return inner
@classmethod
def get(cls, path=None, **kwargs):
return cls.route(path, methods=["GET"], **kwargs)
@classmethod
def post(cls, path=None, **kwargs):
return cls.route(path, methods=["POST"], **kwargs)
@classmethod
def head(cls, path=None, **kwargs):
return cls.route(path, methods=["HEAD"], **kwargs)
@classmethod
def put(cls, path=None, **kwargs):
return cls.route(path, methods=["PUT"], **kwargs)
@classmethod
def delete(cls, path=None, **kwargs):
return cls.route(path, methods=["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):
...
"""
def inner(fn):
route_path = getattr(fn, "_route_path", None)
routes = list(filter(lambda x: x.path == route_path, cls._registered_routes))
for route in routes[:-1]:
cls._registered_routes.remove(route)
return fn
return inner

View File

@@ -0,0 +1,13 @@
from typing import Union, Literal, Callable
from urllib.request import Request
from starlette.middleware import Middleware
from starlette.types import ASGIApp
from starlette.websockets import WebSocket
TRequest = Union[Request, WebSocket]
HTTPMethods = Literal["GET", "POST", "PUT", "PATCH", "DELETE", "OPTIONS"]
PartialMiddleware = Union[
Middleware,
Callable[[ASGIApp], ASGIApp],
]

View File

@@ -0,0 +1,153 @@
import os
from typing import Mapping, Any, Callable
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.routing import Route
from starlette.types import ExceptionHandler
from cpl.api.api_logger import APILogger
from cpl.api.api_settings import ApiSettings
from cpl.api.error import APIError
from cpl.api.middleware.logging import LoggingMiddleware
from cpl.api.middleware.request import RequestMiddleware
from cpl.api.router import Router
from cpl.api.typing import HTTPMethods, PartialMiddleware
from cpl.application.abc.application_abc import ApplicationABC
from cpl.core.configuration import Configuration
from cpl.dependency.service_provider_abc import ServiceProviderABC
_logger = APILogger("API")
class WebApp(ApplicationABC):
def __init__(self, services: ServiceProviderABC):
super().__init__(services)
self._app: Starlette | None = None
self._api_settings = Configuration.get(ApiSettings)
self._routes: list[Route] = []
self._middleware: list[Middleware] = [
Middleware(RequestMiddleware),
Middleware(LoggingMiddleware),
]
self._exception_handlers: Mapping[Any, ExceptionHandler] = {Exception: self.handle_exception}
@staticmethod
async def handle_exception(request: Request, exc: Exception):
if hasattr(request.state, "request_id"):
_logger.error(f"Request {request.state.request_id}", exc)
else:
_logger.error("Request unknown", exc)
if isinstance(exc, APIError):
return JSONResponse({"error": str(exc)}, status_code=exc.status_code)
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 == "":
_logger.warning("No allowed origins specified, allowing all origins")
return ["*"]
_logger.debug(f"Allowed origins: {origins}")
return origins.split(",")
def with_app(self, app: Starlette):
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) -> "WebApp":
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[Route]) -> "WebApp":
self._check_for_app()
assert self._routes is not None, "routes must not be None"
assert all(isinstance(route, Route) for route in routes), "all routes must be of type starlette.routing.Route"
self._routes.extend(routes)
return self
def with_route(self, path: str, fn: Callable[[Request], Any], method: HTTPMethods, **kwargs) -> "WebApp":
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", "POST", "PUT", "DELETE", "PATCH", "OPTIONS", "HEAD"], "method must be a valid HTTP method"
self._routes.append(Route(path, fn, methods=[method], **kwargs))
return self
def with_middleware(self, middleware: PartialMiddleware) -> "WebApp":
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 main(self):
_logger.debug(f"Preparing API")
if self._app is None:
routes = [
Route(
path=route.path,
endpoint=self._services.inject(route.endpoint),
methods=route.methods,
name=route.name,
)
for route in self._routes + Router.get_routes()
]
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
_logger.info(f"Start API on {self._api_settings.host}:{self._api_settings.port}")
uvicorn.run(
app,
host=self._api_settings.host,
port=self._api_settings.port,
log_config=None,
)
_logger.info("Shutdown API")

View File

@@ -0,0 +1,30 @@
[build-system]
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
build-backend = "setuptools.build_meta"
[project]
name = "cpl-api"
version = "2024.7.0"
description = "CPL api"
readme ="CPL api package"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
]
keywords = ["cpl", "api", "backend", "shared", "library"]
dynamic = ["dependencies", "optional-dependencies"]
[project.urls]
Homepage = "https://www.sh-edraft.de"
[tool.setuptools.packages.find]
where = ["."]
include = ["cpl*"]
[tool.setuptools.dynamic]
dependencies = { file = ["requirements.txt"] }
optional-dependencies.dev = { file = ["requirements.dev.txt"] }

View File

@@ -0,0 +1 @@
black==25.1.0

View File

@@ -0,0 +1,6 @@
cpl-auth
cpl-application
cpl-core
cpl-dependency
starlette==0.48.0
python-multipart==0.0.20

View File

@@ -1,6 +1 @@
from .application_abc import ApplicationABC
from .application_builder import ApplicationBuilder
from .application_builder_abc import ApplicationBuilderABC
from .application_extension_abc import ApplicationExtensionABC
from .startup_abc import StartupABC
from .startup_extension_abc import StartupExtensionABC

View File

@@ -0,0 +1,4 @@
from .application_abc import ApplicationABC
from .application_extension_abc import ApplicationExtensionABC
from .startup_abc import StartupABC
from .startup_extension_abc import StartupExtensionABC

View File

@@ -0,0 +1,86 @@
from abc import ABC, abstractmethod
from typing import Callable, Self
from cpl.application.host import Host
from cpl.core.console.console import Console
from cpl.core.log import LogSettings
from cpl.core.log.log_level import LogLevel
from cpl.core.log.logger_abc import LoggerABC
from cpl.dependency.service_provider_abc import ServiceProviderABC
def __not_implemented__(package: str, func: Callable):
raise NotImplementedError(f"Package {package} is required to use {func.__name__} method")
class ApplicationABC(ABC):
r"""ABC for the Application class
Parameters:
services: :class:`cpl.dependency.service_provider_abc.ServiceProviderABC`
Contains instances of prepared objects
"""
@abstractmethod
def __init__(self, services: ServiceProviderABC):
self._services = services
@classmethod
def extend(cls, name: str | Callable, func: Callable[[Self], Self]):
r"""Extend the Application with a custom method
Parameters:
name: :class:`str`
Name of the method
func: :class:`Callable[[Self], Self]`
Function that takes the Application as a parameter and returns it
"""
if callable(name):
name = name.__name__
setattr(cls, name, func)
return cls
def with_logging(self, level: LogLevel = None):
if level is None:
from cpl.core.configuration.configuration import Configuration
settings = Configuration.get(LogSettings)
level = settings.level if settings else LogLevel.info
logger = self._services.get_service(LoggerABC)
logger.set_level(level)
def with_permissions(self, *args, **kwargs):
__not_implemented__("cpl-auth", self.with_permissions)
def with_migrations(self, *args, **kwargs):
__not_implemented__("cpl-database", self.with_migrations)
def with_seeders(self, *args, **kwargs):
__not_implemented__("cpl-database", self.with_seeders)
def with_extension(self, func: Callable[[Self, ...], None], *args, **kwargs):
r"""Extend the Application with a custom method
Parameters:
func: :class:`Callable[[Self], Self]`
Function that takes the Application as a parameter and returns it
"""
assert func is not None, "func must not be None"
assert callable(func), "func must be callable"
func(self, *args, **kwargs)
def run(self):
r"""Entry point
Called by custom Application.main
"""
try:
Host.run(self.main)
except KeyboardInterrupt:
Console.close()
@abstractmethod
def main(self): ...

View File

@@ -0,0 +1,10 @@
from abc import ABC, abstractmethod
from cpl.dependency import ServiceProviderABC
class ApplicationExtensionABC(ABC):
@staticmethod
@abstractmethod
def run(services: ServiceProviderABC): ...

View File

@@ -0,0 +1,21 @@
from abc import ABC, abstractmethod
from cpl.dependency.service_collection import ServiceCollection
class StartupABC(ABC):
r"""ABC for the startup class"""
@staticmethod
@abstractmethod
def configure_configuration():
r"""Creates configuration of application"""
@staticmethod
@abstractmethod
def configure_services(service: ServiceCollection):
r"""Creates service provider
Parameter:
services: :class:`cpl.dependency.service_collection`
"""

View File

@@ -0,0 +1,20 @@
from abc import ABC, abstractmethod
from cpl.dependency import ServiceCollection
class StartupExtensionABC(ABC):
r"""ABC for startup extension classes"""
@staticmethod
@abstractmethod
def configure_configuration():
r"""Creates configuration of application"""
@staticmethod
@abstractmethod
def configure_services(services: ServiceCollection):
r"""Creates service provider
Parameter:
services: :class:`cpl.dependency.service_collection`
"""

View File

@@ -1,57 +0,0 @@
from abc import ABC, abstractmethod
from typing import Optional
from cpl.dependency.service_provider_abc import ServiceProviderABC
from cpl.core.console.console import Console
class ApplicationABC(ABC):
r"""ABC for the Application class
Parameters:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
Contains object loaded from appsettings
services: :class:`cpl.dependency.service_provider_abc.ServiceProviderABC`
Contains instances of prepared objects
"""
@abstractmethod
def __init__(self, services: ServiceProviderABC):
self._services: Optional[ServiceProviderABC] = services
def run(self):
r"""Entry point
Called by custom Application.main
"""
try:
self.configure()
self.main()
except KeyboardInterrupt:
Console.close()
async def run_async(self):
r"""Entry point
Called by custom Application.main
"""
try:
await self.configure()
await self.main()
except KeyboardInterrupt:
Console.close()
@abstractmethod
def configure(self):
r"""Configure the application
Called by :class:`cpl.application.application_abc.ApplicationABC.run`
"""
@abstractmethod
def main(self):
r"""Custom entry point
Called by :class:`cpl.application.application_abc.ApplicationABC.run`
"""

View File

@@ -1,97 +1,65 @@
from typing import Type, Optional, Callable, Union
import asyncio
from typing import Type, Optional, TypeVar, Generic
from cpl.application.application_abc import ApplicationABC
from cpl.application.application_builder_abc import ApplicationBuilderABC
from cpl.application.application_extension_abc import ApplicationExtensionABC
from cpl.application.async_application_extension_abc import AsyncApplicationExtensionABC
from cpl.application.async_startup_abc import AsyncStartupABC
from cpl.application.async_startup_extension_abc import AsyncStartupExtensionABC
from cpl.application.startup_abc import StartupABC
from cpl.application.startup_extension_abc import StartupExtensionABC
from cpl.core.configuration.configuration import Configuration
from cpl.application.abc.application_abc import ApplicationABC
from cpl.application.abc.application_extension_abc import ApplicationExtensionABC
from cpl.application.abc.startup_abc import StartupABC
from cpl.application.abc.startup_extension_abc import StartupExtensionABC
from cpl.application.host import Host
from cpl.dependency.service_collection import ServiceCollection
from cpl.core.environment import Environment
TApp = TypeVar("TApp", bound=ApplicationABC)
class ApplicationBuilder(ApplicationBuilderABC):
r"""This is class is used to build an object of :class:`cpl.application.application_abc.ApplicationABC`
Parameter:
app: Type[:class:`cpl.application.application_abc.ApplicationABC`]
Application to build
"""
class ApplicationBuilder(Generic[TApp]):
def __init__(self, app: Type[ApplicationABC]):
ApplicationBuilderABC.__init__(self)
self._app = app
self._startup: Optional[StartupABC | AsyncStartupABC] = None
assert app is not None, "app must not be None"
assert issubclass(app, ApplicationABC), "app must be an subclass of ApplicationABC or its subclass"
self._app = app if app is not None else ApplicationABC
self._services = ServiceCollection()
self._app_extensions: list[Type[ApplicationExtensionABC | AsyncApplicationExtensionABC]] = []
self._startup_extensions: list[Type[StartupExtensionABC | AsyncStartupABC]] = []
self._startup: Optional[StartupABC] = None
self._app_extensions: list[Type[ApplicationExtensionABC]] = []
self._startup_extensions: list[Type[StartupExtensionABC]] = []
def use_startup(self, startup: Type[StartupABC | AsyncStartupABC]) -> "ApplicationBuilder":
self._startup = startup()
self._async_loop = asyncio.get_event_loop()
@property
def services(self) -> ServiceCollection:
return self._services
@property
def service_provider(self):
return self._services.build()
def with_startup(self, startup: Type[StartupABC]) -> "ApplicationBuilder":
self._startup = startup
return self
def use_extension(
def with_extension(
self,
extension: Type[
ApplicationExtensionABC | AsyncApplicationExtensionABC | StartupExtensionABC | AsyncStartupExtensionABC
],
extension: Type[ApplicationExtensionABC | StartupExtensionABC],
) -> "ApplicationBuilder":
if (
issubclass(extension, ApplicationExtensionABC) or issubclass(extension, AsyncApplicationExtensionABC)
) and extension not in self._app_extensions:
if (issubclass(extension, ApplicationExtensionABC)) and extension not in self._app_extensions:
self._app_extensions.append(extension)
elif (
issubclass(extension, StartupExtensionABC) or issubclass(extension, AsyncStartupExtensionABC)
) and extension not in self._startup_extensions:
elif (issubclass(extension, StartupExtensionABC)) and extension not in self._startup_extensions:
self._startup_extensions.append(extension)
return self
def _build_startup(self):
for ex in self._startup_extensions:
extension = ex()
extension.configure_configuration(Configuration, Environment)
extension.configure_services(self._services, Environment)
def build(self) -> TApp:
for extension in self._startup_extensions:
Host.run(extension.configure_configuration)
Host.run(extension.configure_services, self._services)
if self._startup is not None:
self._startup.configure_configuration(Configuration, Environment)
self._startup.configure_services(self._services, Environment)
Host.run(self._startup.configure_configuration)
Host.run(self._startup.configure_services, self._services)
async def _build_async_startup(self):
for ex in self._startup_extensions:
extension = ex()
await extension.configure_configuration(Configuration, Environment)
await extension.configure_services(self._services, Environment)
for extension in self._app_extensions:
Host.run(extension.run, self.service_provider)
if self._startup is not None:
await self._startup.configure_configuration(Configuration, Environment)
await self._startup.configure_services(self._services, Environment)
def build(self) -> ApplicationABC:
self._build_startup()
config = Configuration
services = self._services.build_service_provider()
for ex in self._app_extensions:
extension = ex()
extension.run(config, services)
return self._app(services)
async def build_async(self) -> ApplicationABC:
await self._build_async_startup()
config = Configuration
services = self._services.build_service_provider()
for ex in self._app_extensions:
extension = ex()
await extension.run(config, services)
return self._app(services)
return self._app(self.service_provider)

View File

@@ -1,47 +0,0 @@
from abc import ABC, abstractmethod
from typing import Type
from cpl.application.application_abc import ApplicationABC
from cpl.application.startup_abc import StartupABC
class ApplicationBuilderABC(ABC):
r"""ABC for the :class:`cpl.application.application_builder.ApplicationBuilder`"""
@abstractmethod
def __init__(self, *args):
pass
@abstractmethod
def use_startup(self, startup: Type[StartupABC]):
r"""Sets the custom startup class to use
Parameter:
startup: Type[:class:`cpl.application.startup_abc.StartupABC`]
Startup class to use
"""
@abstractmethod
async def use_startup(self, startup: Type[StartupABC]):
r"""Sets the custom startup class to use async
Parameter:
startup: Type[:class:`cpl.application.startup_abc.StartupABC`]
Startup class to use
"""
@abstractmethod
def build(self) -> ApplicationABC:
r"""Creates custom application object
Returns:
Object of :class:`cpl.application.application_abc.ApplicationABC`
"""
@abstractmethod
async def build_async(self) -> ApplicationABC:
r"""Creates custom application object async
Returns:
Object of :class:`cpl.application.application_abc.ApplicationABC`
"""

View File

@@ -1,14 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration import Configuration
from cpl.dependency import ServiceProviderABC
class ApplicationExtensionABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
def run(self, config: Configuration, services: ServiceProviderABC):
pass

View File

@@ -1,14 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration import Configuration
from cpl.dependency import ServiceProviderABC
class AsyncApplicationExtensionABC(ABC):
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def run(self, config: Configuration, services: ServiceProviderABC):
pass

View File

@@ -1,23 +0,0 @@
from abc import ABC, abstractmethod
from cpl.dependency.service_collection_abc import ServiceCollectionABC
class AsyncStartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def configure_configuration(self):
r"""Creates configuration of application"""
@abstractmethod
async def configure_services(self, service: ServiceCollectionABC):
r"""Creates service provider
Parameter:
services: :class:`cpl.dependency.service_collection_abc`
"""

View File

@@ -1,31 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration.configuration import Configuration
from cpl.dependency.service_collection_abc import ServiceCollectionABC
from cpl.core.environment.environment import Environment
class AsyncStartupExtensionABC(ABC):
r"""ABC for startup extension classes"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
async def configure_configuration(self, config: Configuration, env: Environment):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.Configuration`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
async def configure_services(self, service: ServiceCollectionABC, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.dependency.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

View File

@@ -0,0 +1,17 @@
import asyncio
from typing import Callable
class Host:
_loop = asyncio.get_event_loop()
@classmethod
def get_loop(cls):
return cls._loop
@classmethod
def run(cls, func: Callable, *args, **kwargs):
if asyncio.iscoroutinefunction(func):
return cls._loop.run_until_complete(func(*args, **kwargs))
return func(*args, **kwargs)

View File

@@ -1,31 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration import Configuration
from cpl.dependency.service_collection_abc import ServiceCollectionABC
from cpl.core.environment import Environment
class StartupABC(ABC):
r"""ABC for the startup class"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
def configure_configuration(self, config: Configuration, env: Environment):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
def configure_services(self, service: ServiceCollectionABC, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.dependency.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

View File

@@ -1,33 +0,0 @@
from abc import ABC, abstractmethod
from cpl.core.configuration import Configuration
from cpl.dependency.service_collection_abc import ServiceCollectionABC
from cpl.core.environment.environment import Environment
class StartupExtensionABC(ABC):
r"""ABC for startup extension classes"""
@abstractmethod
def __init__(self):
pass
@abstractmethod
def configure_configuration(self, config: Configuration, env: Environment):
r"""Creates configuration of application
Parameter:
config: :class:`cpl.core.configuration.configuration_abc.ConfigurationABC`
env: :class:`cpl.core.environment.application_environment_abc`
"""
@abstractmethod
def configure_services(self, service: ServiceCollectionABC, env: Environment):
r"""Creates service provider
Parameter:
services: :class:`cpl.dependency.service_collection_abc`
env: :class:`cpl.core.environment.application_environment_abc`
"""

View File

@@ -0,0 +1,82 @@
from enum import Enum
from typing import Type
from cpl.application.abc import ApplicationABC as _ApplicationABC
from cpl.auth import permission as _permission
from cpl.auth.keycloak.keycloak_admin import KeycloakAdmin as _KeycloakAdmin
from cpl.auth.keycloak.keycloak_client import KeycloakClient as _KeycloakClient
from cpl.dependency.service_collection import ServiceCollection as _ServiceCollection
from .auth_logger import AuthLogger
from .keycloak_settings import KeycloakSettings
from .permission_seeder import PermissionSeeder
def _with_permissions(self: _ApplicationABC, *permissions: Type[Enum]) -> _ApplicationABC:
from cpl.auth.permission.permissions_registry import PermissionsRegistry
for perm in permissions:
PermissionsRegistry.with_enum(perm)
return self
def _add_daos(collection: _ServiceCollection):
from .schema._administration.auth_user_dao import AuthUserDao
from .schema._administration.api_key_dao import ApiKeyDao
from .schema._permission.api_key_permission_dao import ApiKeyPermissionDao
from .schema._permission.permission_dao import PermissionDao
from .schema._permission.role_dao import RoleDao
from .schema._permission.role_permission_dao import RolePermissionDao
from .schema._permission.role_user_dao import RoleUserDao
collection.add_singleton(AuthUserDao)
collection.add_singleton(ApiKeyDao)
collection.add_singleton(ApiKeyPermissionDao)
collection.add_singleton(PermissionDao)
collection.add_singleton(RoleDao)
collection.add_singleton(RolePermissionDao)
collection.add_singleton(RoleUserDao)
def add_auth(collection: _ServiceCollection):
import os
from cpl.core.console import Console
from cpl.database.service.migration_service import MigrationService
from cpl.database.model.server_type import ServerType, ServerTypes
try:
collection.add_singleton(_KeycloakClient)
collection.add_singleton(_KeycloakAdmin)
_add_daos(collection)
provider = collection.build()
migration_service: MigrationService = provider.get_service(MigrationService)
if ServerType.server_type == ServerTypes.POSTGRES:
migration_service.with_directory(
os.path.join(os.path.dirname(os.path.realpath(__file__)), "scripts/postgres")
)
elif ServerType.server_type == ServerTypes.MYSQL:
migration_service.with_directory(os.path.join(os.path.dirname(os.path.realpath(__file__)), "scripts/mysql"))
except ImportError as e:
Console.error("cpl-auth is not installed", str(e))
def add_permission(collection: _ServiceCollection):
from cpl.auth.permission_seeder import PermissionSeeder
from cpl.database.abc.data_seeder_abc import DataSeederABC
from cpl.auth.permission.permissions_registry import PermissionsRegistry
from cpl.auth.permission.permissions import Permissions
try:
collection.add_singleton(DataSeederABC, PermissionSeeder)
PermissionsRegistry.with_enum(Permissions)
except ImportError as e:
from cpl.core.console import Console
Console.error("cpl-auth is not installed", str(e))
_ServiceCollection.with_module(add_auth, __name__)
_ServiceCollection.with_module(add_permission, _permission.__name__)
_ApplicationABC.extend(_ApplicationABC.with_permissions, _with_permissions)

View File

@@ -0,0 +1,8 @@
from cpl.core.log import Logger
from cpl.core.typing import Source
class AuthLogger(Logger):
def __init__(self, source: Source):
Logger.__init__(self, source, "auth")

View File

@@ -0,0 +1,3 @@
from .keycloak_admin import KeycloakAdmin
from .keycloak_client import KeycloakClient
from .keycloak_user import KeycloakUser

View File

@@ -0,0 +1,24 @@
from keycloak import KeycloakAdmin as _KeycloakAdmin, KeycloakOpenIDConnection
from cpl.auth.auth_logger import AuthLogger
from cpl.auth.keycloak_settings import KeycloakSettings
_logger = AuthLogger("keycloak")
class KeycloakAdmin(_KeycloakAdmin):
def __init__(self, settings: KeycloakSettings):
_logger.info("Initializing Keycloak admin")
_connection = KeycloakOpenIDConnection(
server_url=settings.url,
client_id=settings.client_id,
realm_name=settings.realm,
client_secret_key=settings.client_secret,
)
_KeycloakAdmin.__init__(
self,
connection=_connection,
)
self.__connection = _connection

View File

@@ -0,0 +1,26 @@
from keycloak import KeycloakOpenID, KeycloakAdmin, KeycloakOpenIDConnection
from cpl.auth.auth_logger import AuthLogger
from cpl.auth.keycloak_settings import KeycloakSettings
_logger = AuthLogger("keycloak")
class KeycloakClient(KeycloakOpenID):
def __init__(self, settings: KeycloakSettings):
KeycloakOpenID.__init__(
self,
server_url=settings.url,
client_id=settings.client_id,
realm_name=settings.realm,
client_secret_key=settings.client_secret,
)
_logger.info("Initializing Keycloak client")
connection = KeycloakOpenIDConnection(
server_url=settings.url,
client_id=settings.client_id,
realm_name=settings.realm,
client_secret_key=settings.client_secret,
)
self._admin = KeycloakAdmin(connection=connection)

View File

@@ -0,0 +1,36 @@
from cpl.core.utils.get_value import get_value
from cpl.dependency import ServiceProviderABC
class KeycloakUser:
def __init__(self, source: dict):
self._username = get_value(source, "preferred_username", str)
self._email = get_value(source, "email", str)
self._email_verified = get_value(source, "email_verified", bool)
self._name = get_value(source, "name", str)
@property
def username(self) -> str:
return self._username
@property
def email(self) -> str:
return self._email
@property
def email_verified(self) -> bool:
return self._email_verified
@property
def name(self) -> str:
return self._name
# Attrs from keycloak
@property
def id(self) -> str:
from cpl.auth import KeycloakAdmin
keycloak_admin: KeycloakAdmin = ServiceProviderABC.get_global_service(KeycloakAdmin)
return keycloak_admin.get_user_id(self._username)

View File

@@ -0,0 +1,17 @@
from typing import Optional
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
class KeycloakSettings(ConfigurationModelABC):
def __init__(
self,
src: Optional[dict] = None,
):
ConfigurationModelABC.__init__(self, src, "KEYCLOAK")
self.option("url", str, required=True)
self.option("client_id", str, required=True)
self.option("realm", str, required=True)
self.option("client_secret", str, required=True)

View File

@@ -0,0 +1,36 @@
from enum import Enum
class Permissions(Enum):
""" """
"""
Administration
"""
# administrator
administrator = "administrator"
# api keys
api_keys = "api_keys"
api_keys_create = "api_keys.create"
api_keys_update = "api_keys.update"
api_keys_delete = "api_keys.delete"
# users
users = "users"
users_create = "users.create"
users_update = "users.update"
users_delete = "users.delete"
# settings
settings = "settings"
settings_update = "settings.update"
"""
Permissions
"""
# roles
roles = "roles"
roles_create = "roles.create"
roles_update = "roles.update"
roles_delete = "roles.delete"

View File

@@ -0,0 +1,24 @@
from enum import Enum
from typing import Type
class PermissionsRegistry:
_permissions: dict[str, str] = {}
@classmethod
def get(cls):
return cls._permissions.keys()
@classmethod
def descriptions(cls):
return {x: cls._permissions[x] for x in cls._permissions if cls._permissions[x] is not None}
@classmethod
def set(cls, permission: str, description: str = None):
cls._permissions[permission] = description
@classmethod
def with_enum(cls, e: Type[Enum]):
perms = [x.value for x in e]
for perm in perms:
cls.set(str(perm))

View File

@@ -0,0 +1,120 @@
from cpl.auth.permission.permissions import Permissions
from cpl.auth.permission.permissions_registry import PermissionsRegistry
from cpl.auth.schema import (
Permission,
Role,
RolePermission,
ApiKey,
ApiKeyPermission,
PermissionDao,
RoleDao,
RolePermissionDao,
ApiKeyDao,
ApiKeyPermissionDao,
)
from cpl.core.utils.get_value import get_value
from cpl.database.abc.data_seeder_abc import DataSeederABC
from cpl.database.db_logger import DBLogger
_logger = DBLogger(__name__)
class PermissionSeeder(DataSeederABC):
def __init__(
self,
permission_dao: PermissionDao,
role_dao: RoleDao,
role_permission_dao: RolePermissionDao,
api_key_dao: ApiKeyDao,
api_key_permission_dao: ApiKeyPermissionDao,
):
DataSeederABC.__init__(self)
self._permission_dao = permission_dao
self._role_dao = role_dao
self._role_permission_dao = role_permission_dao
self._api_key_dao = api_key_dao
self._api_key_permission_dao = api_key_permission_dao
async def seed(self):
permissions = await self._permission_dao.get_all()
possible_permissions = [permission for permission in PermissionsRegistry.get()]
if len(permissions) == len(possible_permissions):
_logger.info("Permissions already existing")
await self._update_missing_descriptions()
return
to_delete = []
for permission in permissions:
if permission.name in possible_permissions:
continue
to_delete.append(permission)
await self._permission_dao.delete_many(to_delete, hard_delete=True)
_logger.warning("Permissions incomplete")
permission_names = [permission.name for permission in permissions]
await self._permission_dao.create_many(
[
Permission(
0,
permission,
get_value(PermissionsRegistry.descriptions(), permission, str),
)
for permission in possible_permissions
if permission not in permission_names
]
)
await self._update_missing_descriptions()
await self._add_missing_to_role()
await self._add_missing_to_api_key()
async def _add_missing_to_role(self):
admin_role = await self._role_dao.find_single_by([{Role.id: 1}, {Role.name: "admin"}])
if admin_role is None:
return
admin_permissions = await self._role_permission_dao.get_by_role_id(admin_role.id, with_deleted=True)
to_assign = [
RolePermission(0, admin_role.id, permission.id)
for permission in await self._permission_dao.get_all()
if permission.id not in [x.permission_id for x in admin_permissions]
]
await self._role_permission_dao.create_many(to_assign)
async def _add_missing_to_api_key(self):
admin_api_key = await self._api_key_dao.find_single_by([{ApiKey.id: 1}, {ApiKey.identifier: "admin"}])
if admin_api_key is None:
return
admin_permissions = await self._api_key_permission_dao.find_by_api_key_id(admin_api_key.id, with_deleted=True)
to_assign = [
ApiKeyPermission(0, admin_api_key.id, permission.id)
for permission in await self._permission_dao.get_all()
if permission.id not in [x.permission_id for x in admin_permissions]
]
await self._api_key_permission_dao.create_many(to_assign)
async def _update_missing_descriptions(self):
permissions = {
permission.name: permission
for permission in await self._permission_dao.find_by([{Permission.description: None}])
}
to_update = []
if len(permissions) == 0:
return
for key in PermissionsRegistry.descriptions():
if key.value not in permissions:
continue
permissions[key.value].description = PermissionsRegistry.descriptions()[key]
to_update.append(permissions[key.value])
if len(to_update) == 0:
return
await self._permission_dao.update_many(to_update)

View File

@@ -0,0 +1,15 @@
from ._administration.api_key import ApiKey
from ._administration.api_key_dao import ApiKeyDao
from ._administration.auth_user import AuthUser
from ._administration.auth_user_dao import AuthUserDao
from ._permission.api_key_permission import ApiKeyPermission
from ._permission.api_key_permission_dao import ApiKeyPermissionDao
from ._permission.permission import Permission
from ._permission.permission_dao import PermissionDao
from ._permission.role import Role
from ._permission.role_dao import RoleDao
from ._permission.role_permission import RolePermission
from ._permission.role_permission_dao import RolePermissionDao
from ._permission.role_user import RoleUser
from ._permission.role_user_dao import RoleUserDao

View File

@@ -0,0 +1,66 @@
import secrets
from datetime import datetime
from typing import Optional, Union
from async_property import async_property
from cpl.auth.permission.permissions import Permissions
from cpl.core.environment.environment import Environment
from cpl.core.log.logger import Logger
from cpl.core.typing import Id, SerialId
from cpl.core.utils.credential_manager import CredentialManager
from cpl.database.abc.db_model_abc import DbModelABC
from cpl.dependency.service_provider_abc import ServiceProviderABC
_logger = Logger(__name__)
class ApiKey(DbModelABC):
def __init__(
self,
id: SerialId,
identifier: str,
key: Union[str, bytes],
deleted: bool = False,
editor_id: Optional[Id] = None,
created: Optional[datetime] = None,
updated: Optional[datetime] = None,
):
DbModelABC.__init__(self, id, deleted, editor_id, created, updated)
self._identifier = identifier
self._key = key
@property
def identifier(self) -> str:
return self._identifier
@property
def key(self) -> str:
return self._key
@property
def plain_key(self) -> str:
return CredentialManager.decrypt(self.key)
@async_property
async def permissions(self):
from cpl.auth.schema._permission.api_key_permission_dao import ApiKeyPermissionDao
apiKeyPermissionDao = ServiceProviderABC.get_global_provider().get_service(ApiKeyPermissionDao)
return [await x.permission for x in await apiKeyPermissionDao.find_by_api_key_id(self.id)]
async def has_permission(self, permission: Permissions) -> bool:
return permission.value in [x.name for x in await self.permissions]
def set_new_api_key(self):
self._key = self.new_key()
@staticmethod
def new_key() -> str:
return CredentialManager.encrypt(f"api_{secrets.token_urlsafe(Environment.get("API_KEY_LENGTH", int, 64))}")
@classmethod
def new(cls, identifier: str) -> "ApiKey":
return ApiKey(0, identifier, cls.new_key())

View File

@@ -0,0 +1,32 @@
from typing import Optional
from cpl.auth.schema._administration.api_key import ApiKey
from cpl.database import TableManager
from cpl.database.abc import DbModelDaoABC
from cpl.database.db_logger import DBLogger
_logger = DBLogger(__name__)
class ApiKeyDao(DbModelDaoABC[ApiKey]):
def __init__(self):
DbModelDaoABC.__init__(self, __name__, ApiKey, TableManager.get("api_keys"))
self.attribute(ApiKey.identifier, str)
self.attribute(ApiKey.key, str, "keystring")
async def get_by_identifier(self, ident: str) -> ApiKey:
result = await self._db.select_map(f"SELECT * FROM {self._table_name} WHERE Identifier = '{ident}'")
return self.to_object(result[0])
async def get_by_key(self, key: str) -> ApiKey:
result = await self._db.select_map(f"SELECT * FROM {self._table_name} WHERE Keystring = '{key}'")
return self.to_object(result[0])
async def find_by_key(self, key: str) -> Optional[ApiKey]:
result = await self._db.select_map(f"SELECT * FROM {self._table_name} WHERE Keystring = '{key}'")
if not result or len(result) == 0:
return None
return self.to_object(result[0])

View File

@@ -0,0 +1,89 @@
import uuid
from datetime import datetime
from typing import Optional
from async_property import async_property
from keycloak import KeycloakGetError
from cpl.auth.keycloak import KeycloakAdmin
from cpl.auth.auth_logger import AuthLogger
from cpl.auth.permission.permissions import Permissions
from cpl.core.typing import SerialId
from cpl.database.abc import DbModelABC
from cpl.dependency import ServiceProviderABC
_logger = AuthLogger(__name__)
class AuthUser(DbModelABC):
def __init__(
self,
id: SerialId,
keycloak_id: 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._keycloak_id = keycloak_id
@property
def keycloak_id(self) -> str:
return self._keycloak_id
@property
def username(self):
if self._keycloak_id == str(uuid.UUID(int=0)):
return "ANONYMOUS"
try:
keycloak_admin: KeycloakAdmin = ServiceProviderABC.get_global_service(KeycloakAdmin)
return keycloak_admin.get_user(self._keycloak_id).get("username")
except KeycloakGetError as e:
return "UNKNOWN"
except Exception as e:
_logger.error(f"Failed to get user {self._keycloak_id} from Keycloak", e)
return "UNKNOWN"
@property
def email(self):
if self._keycloak_id == str(uuid.UUID(int=0)):
return "ANONYMOUS"
try:
keycloak_admin: KeycloakAdmin = ServiceProviderABC.get_global_service(KeycloakAdmin)
return keycloak_admin.get_user(self._keycloak_id).get("email")
except KeycloakGetError as e:
return "UNKNOWN"
except Exception as e:
_logger.error(f"Failed to get user {self._keycloak_id} from Keycloak", e)
return "UNKNOWN"
@async_property
async def roles(self):
from cpl.auth.schema._permission.role_user_dao import RoleUserDao
role_user_dao: RoleUserDao = ServiceProviderABC.get_global_service(RoleUserDao)
return [await x.role for x in await role_user_dao.get_by_user_id(self.id)]
@async_property
async def permissions(self):
from cpl.auth.schema._administration.auth_user_dao import AuthUserDao
auth_user_dao: AuthUserDao = ServiceProviderABC.get_global_service(AuthUserDao)
return await auth_user_dao.get_permissions(self.id)
async def has_permission(self, permission: Permissions) -> bool:
from cpl.auth.schema._administration.auth_user_dao import AuthUserDao
auth_user_dao: AuthUserDao = ServiceProviderABC.get_global_service(AuthUserDao)
return await auth_user_dao.has_permission(self.id, permission)
async def anonymize(self):
from cpl.auth.schema._administration.auth_user_dao import AuthUserDao
auth_user_dao: AuthUserDao = ServiceProviderABC.get_global_service(AuthUserDao)
self._keycloak_id = str(uuid.UUID(int=0))
await auth_user_dao.update(self)

View File

@@ -0,0 +1,72 @@
from typing import Optional, Union
from cpl.auth.permission.permissions import Permissions
from cpl.auth.schema._administration.auth_user import AuthUser
from cpl.database import TableManager
from cpl.database.abc import DbModelDaoABC
from cpl.database.db_logger import DBLogger
from cpl.database.external_data_temp_table_builder import ExternalDataTempTableBuilder
from cpl.dependency import ServiceProviderABC
_logger = DBLogger(__name__)
class AuthUserDao(DbModelDaoABC[AuthUser]):
def __init__(self):
DbModelDaoABC.__init__(self, __name__, AuthUser, TableManager.get("auth_users"))
self.attribute(AuthUser.keycloak_id, str, aliases=["keycloakId"])
async def get_users():
return [(x.id, x.username, x.email) for x in await self.get_all()]
self.use_external_fields(
ExternalDataTempTableBuilder()
.with_table_name(self._table_name)
.with_field("id", "int", True)
.with_field("username", "text")
.with_field("email", "text")
.with_value_getter(get_users)
)
async def get_by_keycloak_id(self, keycloak_id: str) -> AuthUser:
return await self.get_single_by({AuthUser.keycloak_id: keycloak_id})
async def find_by_keycloak_id(self, keycloak_id: str) -> Optional[AuthUser]:
return await self.find_single_by({AuthUser.keycloak_id: keycloak_id})
async def has_permission(self, user_id: int, permission: Union[Permissions, str]) -> bool:
from cpl.auth.schema._permission.permission_dao import PermissionDao
permission_dao: PermissionDao = ServiceProviderABC.get_global_service(PermissionDao)
p = await permission_dao.get_by_name(permission if isinstance(permission, str) else permission.value)
result = await self._db.select_map(
f"""
SELECT COUNT(*)
FROM permission.role_users ru
JOIN permission.role_permissions rp ON ru.roleId = rp.roleId
WHERE ru.userId = {user_id}
AND rp.permissionId = {p.id}
AND ru.deleted = FALSE
AND rp.deleted = FALSE;
"""
)
if result is None or len(result) == 0:
return False
return result[0]["count"] > 0
async def get_permissions(self, user_id: int) -> list[Permissions]:
result = await self._db.select_map(
f"""
SELECT p.*
FROM permission.permissions p
JOIN permission.role_permissions rp ON p.id = rp.permissionId
JOIN permission.role_users ru ON rp.roleId = ru.roleId
WHERE ru.userId = {user_id}
AND rp.deleted = FALSE
AND ru.deleted = FALSE;
"""
)
return [Permissions(p["name"]) for p in result]

View File

@@ -0,0 +1,46 @@
from datetime import datetime
from typing import Optional
from async_property import async_property
from cpl.core.typing import SerialId
from cpl.database.abc import DbJoinModelABC
from cpl.dependency import ServiceProviderABC
class ApiKeyPermission(DbJoinModelABC):
def __init__(
self,
id: SerialId,
api_key_id: SerialId,
permission_id: SerialId,
deleted: bool = False,
editor_id: Optional[SerialId] = None,
created: Optional[datetime] = None,
updated: Optional[datetime] = None,
):
DbJoinModelABC.__init__(self, api_key_id, permission_id, id, deleted, editor_id, created, updated)
self._api_key_id = api_key_id
self._permission_id = permission_id
@property
def api_key_id(self) -> int:
return self._api_key_id
@async_property
async def api_key(self):
from cpl.auth.schema._administration.api_key_dao import ApiKeyDao
api_key_dao: ApiKeyDao = ServiceProviderABC.get_global_service(ApiKeyDao)
return await api_key_dao.get_by_id(self._api_key_id)
@property
def permission_id(self) -> int:
return self._permission_id
@async_property
async def permission(self):
from cpl.auth.schema._permission.permission_dao import PermissionDao
permission_dao: PermissionDao = ServiceProviderABC.get_global_service(PermissionDao)
return await permission_dao.get_by_id(self._permission_id)

View File

@@ -0,0 +1,29 @@
from cpl.auth.schema._permission.api_key_permission import ApiKeyPermission
from cpl.database import TableManager
from cpl.database.abc import DbModelDaoABC
from cpl.database.db_logger import DBLogger
_logger = DBLogger(__name__)
class ApiKeyPermissionDao(DbModelDaoABC[ApiKeyPermission]):
def __init__(self):
DbModelDaoABC.__init__(self, __name__, ApiKeyPermission, TableManager.get("api_key_permissions"))
self.attribute(ApiKeyPermission.api_key_id, int)
self.attribute(ApiKeyPermission.permission_id, int)
async def find_by_api_key_id(self, api_key_id: int, with_deleted=False) -> list[ApiKeyPermission]:
f = [{ApiKeyPermission.api_key_id: api_key_id}]
if not with_deleted:
f.append({ApiKeyPermission.deleted: False})
return await self.find_by(f)
async def find_by_permission_id(self, permission_id: int, with_deleted=False) -> list[ApiKeyPermission]:
f = [{ApiKeyPermission.permission_id: permission_id}]
if not with_deleted:
f.append({ApiKeyPermission.deleted: False})
return await self.find_by(f)

View File

@@ -0,0 +1,37 @@
from datetime import datetime
from typing import Optional
from cpl.core.typing import SerialId
from cpl.database.abc import DbModelABC
class Permission(DbModelABC):
def __init__(
self,
id: SerialId,
name: str,
description: 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._description = description
@property
def name(self) -> str:
return self._name
@name.setter
def name(self, value: str):
self._name = value
@property
def description(self) -> str:
return self._description
@description.setter
def description(self, value: str):
self._description = value

View File

@@ -0,0 +1,21 @@
from typing import Optional
from cpl.auth.schema._permission.permission import Permission
from cpl.database import TableManager
from cpl.database.abc import DbModelDaoABC
from cpl.database.db_logger import DBLogger
_logger = DBLogger(__name__)
class PermissionDao(DbModelDaoABC[Permission]):
def __init__(self):
DbModelDaoABC.__init__(self, __name__, Permission, TableManager.get("permissions"))
self.attribute(Permission.name, str)
self.attribute(Permission.description, Optional[str])
async def get_by_name(self, name: str) -> Permission:
result = await self._db.select_map(f"SELECT * FROM {self._table_name} WHERE Name = '{name}'")
return self.to_object(result[0])

View File

@@ -0,0 +1,66 @@
from datetime import datetime
from typing import Optional
from async_property import async_property
from cpl.auth.permission.permissions import Permissions
from cpl.core.typing import SerialId
from cpl.database.abc import DbModelABC
from cpl.dependency import ServiceProviderABC
class Role(DbModelABC):
def __init__(
self,
id: SerialId,
name: str,
description: 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._description = description
@property
def name(self) -> str:
return self._name
@name.setter
def name(self, value: str):
self._name = value
@property
def description(self) -> str:
return self._description
@description.setter
def description(self, value: str):
self._description = value
@async_property
async def permissions(self):
from cpl.auth.schema._permission.role_permission_dao import RolePermissionDao
role_permission_dao: RolePermissionDao = ServiceProviderABC.get_global_service(RolePermissionDao)
return [await x.permission for x in await role_permission_dao.get_by_role_id(self.id)]
@async_property
async def users(self):
from cpl.auth.schema._permission.role_user_dao import RoleUserDao
role_user_dao: RoleUserDao = ServiceProviderABC.get_global_service(RoleUserDao)
return [await x.user for x in await role_user_dao.get_by_role_id(self.id)]
async def has_permission(self, permission: Permissions) -> bool:
from cpl.auth.schema._permission.permission_dao import PermissionDao
from cpl.auth.schema._permission.role_permission_dao import RolePermissionDao
permission_dao: PermissionDao = ServiceProviderABC.get_global_service(PermissionDao)
role_permission_dao: RolePermissionDao = ServiceProviderABC.get_global_service(RolePermissionDao)
p = await permission_dao.get_by_name(permission.value)
return p.id in [x.id for x in await role_permission_dao.get_by_role_id(self.id)]

View File

@@ -0,0 +1,17 @@
from cpl.auth.schema._permission.role import Role
from cpl.database import TableManager
from cpl.database.abc import DbModelDaoABC
from cpl.database.db_logger import DBLogger
_logger = DBLogger(__name__)
class RoleDao(DbModelDaoABC[Role]):
def __init__(self):
DbModelDaoABC.__init__(self, __name__, Role, TableManager.get("roles"))
self.attribute(Role.name, str)
self.attribute(Role.description, str)
async def get_by_name(self, name: str) -> Role:
result = await self._db.select_map(f"SELECT * FROM {self._table_name} WHERE Name = '{name}'")
return self.to_object(result[0])

View File

@@ -0,0 +1,46 @@
from datetime import datetime
from typing import Optional
from async_property import async_property
from cpl.core.typing import SerialId
from cpl.database.abc import DbModelABC
from cpl.dependency import ServiceProviderABC
class RolePermission(DbModelABC):
def __init__(
self,
id: SerialId,
role_id: SerialId,
permission_id: SerialId,
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._role_id = role_id
self._permission_id = permission_id
@property
def role_id(self) -> int:
return self._role_id
@async_property
async def role(self):
from cpl.auth.schema._permission.role_dao import RoleDao
role_dao: RoleDao = ServiceProviderABC.get_global_service(RoleDao)
return await role_dao.get_by_id(self._role_id)
@property
def permission_id(self) -> int:
return self._permission_id
@async_property
async def permission(self):
from cpl.auth.schema._permission.permission_dao import PermissionDao
permission_dao: PermissionDao = ServiceProviderABC.get_global_service(PermissionDao)
return await permission_dao.get_by_id(self._permission_id)

View File

@@ -0,0 +1,29 @@
from cpl.auth.schema._permission.role_permission import RolePermission
from cpl.database import TableManager
from cpl.database.abc import DbModelDaoABC
from cpl.database.db_logger import DBLogger
_logger = DBLogger(__name__)
class RolePermissionDao(DbModelDaoABC[RolePermission]):
def __init__(self):
DbModelDaoABC.__init__(self, __name__, RolePermission, TableManager.get("role_permissions"))
self.attribute(RolePermission.role_id, int)
self.attribute(RolePermission.permission_id, int)
async def get_by_role_id(self, role_id: int, with_deleted=False) -> list[RolePermission]:
f = [{RolePermission.role_id: role_id}]
if not with_deleted:
f.append({RolePermission.deleted: False})
return await self.find_by(f)
async def get_by_permission_id(self, permission_id: int, with_deleted=False) -> list[RolePermission]:
f = [{RolePermission.permission_id: permission_id}]
if not with_deleted:
f.append({RolePermission.deleted: False})
return await self.find_by(f)

View File

@@ -0,0 +1,46 @@
from datetime import datetime
from typing import Optional
from async_property import async_property
from cpl.core.typing import SerialId
from cpl.database.abc import DbJoinModelABC
from cpl.dependency import ServiceProviderABC
class RoleUser(DbJoinModelABC):
def __init__(
self,
id: SerialId,
user_id: SerialId,
role_id: SerialId,
deleted: bool = False,
editor_id: Optional[SerialId] = None,
created: Optional[datetime] = None,
updated: Optional[datetime] = None,
):
DbJoinModelABC.__init__(self, id, user_id, role_id, deleted, editor_id, created, updated)
self._user_id = user_id
self._role_id = role_id
@property
def user_id(self) -> int:
return self._user_id
@async_property
async def user(self):
from cpl.auth.schema._administration.auth_user_dao import AuthUserDao
auth_user_dao: AuthUserDao = ServiceProviderABC.get_global_service(AuthUserDao)
return await auth_user_dao.get_by_id(self._user_id)
@property
def role_id(self) -> int:
return self._role_id
@async_property
async def role(self):
from cpl.auth.schema._permission.role_dao import RoleDao
role_dao: RoleDao = ServiceProviderABC.get_global_service(RoleDao)
return await role_dao.get_by_id(self._role_id)

View File

@@ -0,0 +1,29 @@
from cpl.auth.schema._permission.role_user import RoleUser
from cpl.database import TableManager
from cpl.database.abc import DbModelDaoABC
from cpl.database.db_logger import DBLogger
_logger = DBLogger(__name__)
class RoleUserDao(DbModelDaoABC[RoleUser]):
def __init__(self):
DbModelDaoABC.__init__(self, __name__, RoleUser, TableManager.get("role_users"))
self.attribute(RoleUser.role_id, int)
self.attribute(RoleUser.user_id, int)
async def get_by_role_id(self, rid: int, with_deleted=False) -> list[RoleUser]:
f = [{RoleUser.role_id: rid}]
if not with_deleted:
f.append({RoleUser.deleted: False})
return await self.find_by(f)
async def get_by_user_id(self, uid: int, with_deleted=False) -> list[RoleUser]:
f = [{RoleUser.user_id: uid}]
if not with_deleted:
f.append({RoleUser.deleted: False})
return await self.find_by(f)

View File

@@ -0,0 +1,44 @@
CREATE TABLE IF NOT EXISTS administration_auth_users
(
id INT AUTO_INCREMENT PRIMARY KEY,
keycloakId CHAR(36) NOT NULL,
-- for history
deleted BOOL 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,
CONSTRAINT UC_KeycloakId UNIQUE (keycloakId),
CONSTRAINT FK_EditorId FOREIGN KEY (editorId) REFERENCES administration_auth_users (id)
);
CREATE TABLE IF NOT EXISTS administration_auth_users_history
(
id INT AUTO_INCREMENT PRIMARY KEY,
keycloakId CHAR(36) NOT NULL,
-- for history
deleted BOOL NOT NULL,
editorId INT NULL,
created TIMESTAMP NOT NULL,
updated TIMESTAMP NOT NULL
);
CREATE TRIGGER TR_administration_auth_usersUpdate
AFTER UPDATE
ON administration_auth_users
FOR EACH ROW
BEGIN
INSERT INTO administration_auth_users_history
(id, keycloakId, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.keycloakId, OLD.deleted, OLD.editorId, OLD.created, NOW());
END;
CREATE TRIGGER TR_administration_auth_usersDelete
AFTER DELETE
ON administration_auth_users
FOR EACH ROW
BEGIN
INSERT INTO administration_auth_users_history
(id, keycloakId, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.keycloakId, 1, OLD.editorId, OLD.created, NOW());
END;

View File

@@ -0,0 +1,46 @@
CREATE TABLE IF NOT EXISTS administration_api_keys
(
id INT AUTO_INCREMENT PRIMARY KEY,
identifier VARCHAR(255) NOT NULL,
keyString VARCHAR(255) NOT NULL,
deleted BOOL 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,
CONSTRAINT UC_Identifier_Key UNIQUE (identifier, keyString),
CONSTRAINT UC_Key UNIQUE (keyString),
CONSTRAINT FK_ApiKeys_Editor FOREIGN KEY (editorId) REFERENCES administration_auth_users (id)
);
CREATE TABLE IF NOT EXISTS administration_api_keys_history
(
id INT AUTO_INCREMENT PRIMARY KEY,
identifier VARCHAR(255) NOT NULL,
keyString VARCHAR(255) NOT NULL,
deleted BOOL NOT NULL,
editorId INT NULL,
created TIMESTAMP NOT NULL,
updated TIMESTAMP NOT NULL
);
CREATE TRIGGER TR_ApiKeysUpdate
AFTER UPDATE
ON administration_api_keys
FOR EACH ROW
BEGIN
INSERT INTO administration_api_keys_history
(id, identifier, keyString, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.identifier, OLD.keyString, OLD.deleted, OLD.editorId, OLD.created, NOW());
END;
CREATE TRIGGER TR_ApiKeysDelete
AFTER DELETE
ON administration_api_keys
FOR EACH ROW
BEGIN
INSERT INTO administration_api_keys_history
(id, identifier, keyString, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.identifier, OLD.keyString, 1, OLD.editorId, OLD.created, NOW());
END;

View File

@@ -0,0 +1,179 @@
CREATE TABLE IF NOT EXISTS permission_permissions
(
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NULL,
deleted BOOL 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,
CONSTRAINT UQ_PermissionName UNIQUE (name),
CONSTRAINT FK_Permissions_Editor FOREIGN KEY (editorId) REFERENCES administration_auth_users (id)
);
CREATE TABLE IF NOT EXISTS permission_permissions_history
(
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NULL,
deleted BOOL NOT NULL,
editorId INT NULL,
created TIMESTAMP NOT NULL,
updated TIMESTAMP NOT NULL
);
CREATE TRIGGER TR_PermissionsUpdate
AFTER UPDATE
ON permission_permissions
FOR EACH ROW
BEGIN
INSERT INTO permission_permissions_history
(id, name, description, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.name, OLD.description, OLD.deleted, OLD.editorId, OLD.created, NOW());
END;
CREATE TRIGGER TR_PermissionsDelete
AFTER DELETE
ON permission_permissions
FOR EACH ROW
BEGIN
INSERT INTO permission_permissions_history
(id, name, description, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.name, OLD.description, 1, OLD.editorId, OLD.created, NOW());
END;
CREATE TABLE IF NOT EXISTS permission_roles
(
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NULL,
deleted BOOL 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,
CONSTRAINT UQ_RoleName UNIQUE (name),
CONSTRAINT FK_Roles_Editor FOREIGN KEY (editorId) REFERENCES administration_auth_users (id)
);
CREATE TABLE IF NOT EXISTS permission_roles_history
(
id INT AUTO_INCREMENT PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NULL,
deleted BOOL NOT NULL,
editorId INT NULL,
created TIMESTAMP NOT NULL,
updated TIMESTAMP NOT NULL
);
CREATE TRIGGER TR_RolesUpdate
AFTER UPDATE
ON permission_roles
FOR EACH ROW
BEGIN
INSERT INTO permission_roles_history
(id, name, description, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.name, OLD.description, OLD.deleted, OLD.editorId, OLD.created, NOW());
END;
CREATE TRIGGER TR_RolesDelete
AFTER DELETE
ON permission_roles
FOR EACH ROW
BEGIN
INSERT INTO permission_roles_history
(id, name, description, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.name, OLD.description, 1, OLD.editorId, OLD.created, NOW());
END;
CREATE TABLE IF NOT EXISTS permission_role_permissions
(
id INT AUTO_INCREMENT PRIMARY KEY,
RoleId INT NOT NULL,
permissionId INT NOT NULL,
deleted BOOL 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,
CONSTRAINT UQ_RolePermission UNIQUE (RoleId, permissionId),
CONSTRAINT FK_RolePermissions_Role FOREIGN KEY (RoleId) REFERENCES permission_roles (id) ON DELETE CASCADE,
CONSTRAINT FK_RolePermissions_Permission FOREIGN KEY (permissionId) REFERENCES permission_permissions (id) ON DELETE CASCADE,
CONSTRAINT FK_RolePermissions_Editor FOREIGN KEY (editorId) REFERENCES administration_auth_users (id)
);
CREATE TABLE IF NOT EXISTS permission_role_permissions_history
(
id INT AUTO_INCREMENT PRIMARY KEY,
RoleId INT NOT NULL,
permissionId INT NOT NULL,
deleted BOOL NOT NULL,
editorId INT NULL,
created TIMESTAMP NOT NULL,
updated TIMESTAMP NOT NULL
);
CREATE TRIGGER TR_RolePermissionsUpdate
AFTER UPDATE
ON permission_role_permissions
FOR EACH ROW
BEGIN
INSERT INTO permission_role_permissions_history
(id, RoleId, permissionId, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.RoleId, OLD.permissionId, OLD.deleted, OLD.editorId, OLD.created, NOW());
END;
CREATE TRIGGER TR_RolePermissionsDelete
AFTER DELETE
ON permission_role_permissions
FOR EACH ROW
BEGIN
INSERT INTO permission_role_permissions_history
(id, RoleId, permissionId, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.RoleId, OLD.permissionId, 1, OLD.editorId, OLD.created, NOW());
END;
CREATE TABLE IF NOT EXISTS permission_role_auth_users
(
id INT AUTO_INCREMENT PRIMARY KEY,
RoleId INT NOT NULL,
UserId INT NOT NULL,
deleted BOOL 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,
CONSTRAINT UQ_RoleUser UNIQUE (RoleId, UserId),
CONSTRAINT FK_Roleauth_users_Role FOREIGN KEY (RoleId) REFERENCES permission_roles (id) ON DELETE CASCADE,
CONSTRAINT FK_Roleauth_users_User FOREIGN KEY (UserId) REFERENCES administration_auth_users (id) ON DELETE CASCADE,
CONSTRAINT FK_Roleauth_users_Editor FOREIGN KEY (editorId) REFERENCES administration_auth_users (id)
);
CREATE TABLE IF NOT EXISTS permission_role_auth_users_history
(
id INT AUTO_INCREMENT PRIMARY KEY,
RoleId INT NOT NULL,
UserId INT NOT NULL,
deleted BOOL NOT NULL,
editorId INT NULL,
created TIMESTAMP NOT NULL,
updated TIMESTAMP NOT NULL
);
CREATE TRIGGER TR_Roleauth_usersUpdate
AFTER UPDATE
ON permission_role_auth_users
FOR EACH ROW
BEGIN
INSERT INTO permission_role_auth_users_history
(id, RoleId, UserId, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.RoleId, OLD.UserId, OLD.deleted, OLD.editorId, OLD.created, NOW());
END;
CREATE TRIGGER TR_Roleauth_usersDelete
AFTER DELETE
ON permission_role_auth_users
FOR EACH ROW
BEGIN
INSERT INTO permission_role_auth_users_history
(id, RoleId, UserId, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.RoleId, OLD.UserId, 1, OLD.editorId, OLD.created, NOW());
END;

View File

@@ -0,0 +1,46 @@
CREATE TABLE IF NOT EXISTS permission_api_key_permissions
(
id INT AUTO_INCREMENT PRIMARY KEY,
apiKeyId INT NOT NULL,
permissionId INT NOT NULL,
deleted BOOL 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,
CONSTRAINT UQ_ApiKeyPermission UNIQUE (apiKeyId, permissionId),
CONSTRAINT FK_ApiKeyPermissions_ApiKey FOREIGN KEY (apiKeyId) REFERENCES administration_api_keys (id) ON DELETE CASCADE,
CONSTRAINT FK_ApiKeyPermissions_Permission FOREIGN KEY (permissionId) REFERENCES permission_permissions (id) ON DELETE CASCADE,
CONSTRAINT FK_ApiKeyPermissions_Editor FOREIGN KEY (editorId) REFERENCES administration_auth_users (id)
);
CREATE TABLE IF NOT EXISTS permission_api_key_permissions_history
(
id INT AUTO_INCREMENT PRIMARY KEY,
apiKeyId INT NOT NULL,
permissionId INT NOT NULL,
deleted BOOL NOT NULL,
editorId INT NULL,
created TIMESTAMP NOT NULL,
updated TIMESTAMP NOT NULL
);
CREATE TRIGGER TR_ApiKeyPermissionsUpdate
AFTER UPDATE
ON permission_api_key_permissions
FOR EACH ROW
BEGIN
INSERT INTO permission_api_key_permissions_history
(id, apiKeyId, permissionId, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.apiKeyId, OLD.permissionId, OLD.deleted, OLD.editorId, OLD.created, NOW());
END;
CREATE TRIGGER TR_ApiKeyPermissionsDelete
AFTER DELETE
ON permission_api_key_permissions
FOR EACH ROW
BEGIN
INSERT INTO permission_api_key_permissions_history
(id, apiKeyId, permissionId, deleted, editorId, created, updated)
VALUES (OLD.id, OLD.apiKeyId, OLD.permissionId, 1, OLD.editorId, OLD.created, NOW());
END;

View File

@@ -0,0 +1,26 @@
CREATE SCHEMA IF NOT EXISTS administration;
CREATE TABLE IF NOT EXISTS administration.auth_users
(
id SERIAL PRIMARY KEY,
keycloakId UUID NOT NULL,
-- for history
deleted BOOLEAN NOT NULL DEFAULT FALSE,
editorId INT NULL REFERENCES administration.auth_users (id),
created timestamptz NOT NULL DEFAULT NOW(),
updated timestamptz NOT NULL DEFAULT NOW(),
CONSTRAINT UC_KeycloakId UNIQUE (keycloakId)
);
CREATE TABLE IF NOT EXISTS administration.auth_users_history
(
LIKE administration.auth_users
);
CREATE TRIGGER users_history_trigger
BEFORE INSERT OR UPDATE OR DELETE
ON administration.auth_users
FOR EACH ROW
EXECUTE FUNCTION public.history_trigger_function();

View File

@@ -0,0 +1,28 @@
CREATE SCHEMA IF NOT EXISTS administration;
CREATE TABLE IF NOT EXISTS administration.api_keys
(
id SERIAL PRIMARY KEY,
identifier VARCHAR(255) NOT NULL,
keyString VARCHAR(255) NOT NULL,
-- for history
deleted BOOLEAN NOT NULL DEFAULT FALSE,
editorId INT NULL REFERENCES administration.auth_users (id),
created timestamptz NOT NULL DEFAULT NOW(),
updated timestamptz NOT NULL DEFAULT NOW(),
CONSTRAINT UC_Identifier_Key UNIQUE (identifier, keyString),
CONSTRAINT UC_Key UNIQUE (keyString)
);
CREATE TABLE IF NOT EXISTS administration.api_keys_history
(
LIKE administration.api_keys
);
CREATE TRIGGER api_keys_history_trigger
BEFORE INSERT OR UPDATE OR DELETE
ON administration.api_keys
FOR EACH ROW
EXECUTE FUNCTION public.history_trigger_function();

View File

@@ -0,0 +1,105 @@
CREATE SCHEMA IF NOT EXISTS permission;
-- Permissions
CREATE TABLE permission.permissions
(
id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NULL,
-- for history
deleted BOOLEAN NOT NULL DEFAULT FALSE,
editorId INT NULL REFERENCES administration.auth_users (id),
created timestamptz NOT NULL DEFAULT NOW(),
updated timestamptz NOT NULL DEFAULT NOW(),
CONSTRAINT UQ_PermissionName UNIQUE (name)
);
CREATE TABLE permission.permissions_history
(
LIKE permission.permissions
);
CREATE TRIGGER versioning_trigger
BEFORE INSERT OR UPDATE OR DELETE
ON permission.permissions
FOR EACH ROW
EXECUTE PROCEDURE public.history_trigger_function();
-- Roles
CREATE TABLE permission.roles
(
id SERIAL PRIMARY KEY,
name VARCHAR(255) NOT NULL,
description TEXT NULL,
-- for history
deleted BOOLEAN NOT NULL DEFAULT FALSE,
editorId INT NULL REFERENCES administration.auth_users (id),
created timestamptz NOT NULL DEFAULT NOW(),
updated timestamptz NOT NULL DEFAULT NOW(),
CONSTRAINT UQ_RoleName UNIQUE (name)
);
CREATE TABLE permission.roles_history
(
LIKE permission.roles
);
CREATE TRIGGER versioning_trigger
BEFORE INSERT OR UPDATE OR DELETE
ON permission.roles
FOR EACH ROW
EXECUTE PROCEDURE public.history_trigger_function();
-- Role permissions
CREATE TABLE permission.role_permissions
(
id SERIAL PRIMARY KEY,
RoleId INT NOT NULL REFERENCES permission.roles (id) ON DELETE CASCADE,
permissionId INT NOT NULL REFERENCES permission.permissions (id) ON DELETE CASCADE,
-- for history
deleted BOOLEAN NOT NULL DEFAULT FALSE,
editorId INT NULL REFERENCES administration.auth_users (id),
created timestamptz NOT NULL DEFAULT NOW(),
updated timestamptz NOT NULL DEFAULT NOW(),
CONSTRAINT UQ_RolePermission UNIQUE (RoleId, permissionId)
);
CREATE TABLE permission.role_permissions_history
(
LIKE permission.role_permissions
);
CREATE TRIGGER versioning_trigger
BEFORE INSERT OR UPDATE OR DELETE
ON permission.role_permissions
FOR EACH ROW
EXECUTE PROCEDURE public.history_trigger_function();
-- Role user
CREATE TABLE permission.role_users
(
id SERIAL PRIMARY KEY,
RoleId INT NOT NULL REFERENCES permission.roles (id) ON DELETE CASCADE,
UserId INT NOT NULL REFERENCES administration.auth_users (id) ON DELETE CASCADE,
-- for history
deleted BOOLEAN NOT NULL DEFAULT FALSE,
editorId INT NULL REFERENCES administration.auth_users (id),
created timestamptz NOT NULL DEFAULT NOW(),
updated timestamptz NOT NULL DEFAULT NOW(),
CONSTRAINT UQ_RoleUser UNIQUE (RoleId, UserId)
);
CREATE TABLE permission.role_users_history
(
LIKE permission.role_users
);
CREATE TRIGGER versioning_trigger
BEFORE INSERT OR UPDATE OR DELETE
ON permission.role_users
FOR EACH ROW
EXECUTE PROCEDURE public.history_trigger_function();

View File

@@ -0,0 +1,24 @@
CREATE TABLE permission.api_key_permissions
(
id SERIAL PRIMARY KEY,
apiKeyId INT NOT NULL REFERENCES administration.api_keys (id) ON DELETE CASCADE,
permissionId INT NOT NULL REFERENCES permission.permissions (id) ON DELETE CASCADE,
-- for history
deleted BOOLEAN NOT NULL DEFAULT FALSE,
editorId INT NULL REFERENCES administration.auth_users (id),
created timestamptz NOT NULL DEFAULT NOW(),
updated timestamptz NOT NULL DEFAULT NOW(),
CONSTRAINT UQ_ApiKeyPermission UNIQUE (apiKeyId, permissionId)
);
CREATE TABLE permission.api_key_permissions_history
(
LIKE permission.api_key_permissions
);
CREATE TRIGGER versioning_trigger
BEFORE INSERT OR UPDATE OR DELETE
ON permission.api_key_permissions
FOR EACH ROW
EXECUTE PROCEDURE public.history_trigger_function();

View File

@@ -0,0 +1,30 @@
[build-system]
requires = ["setuptools>=70.1.0", "wheel>=0.43.0"]
build-backend = "setuptools.build_meta"
[project]
name = "cpl-auth"
version = "2024.7.0"
description = "CPL auth"
readme ="CPL auth package"
requires-python = ">=3.12"
license = { text = "MIT" }
authors = [
{ name = "Sven Heidemann", email = "sven.heidemann@sh-edraft.de" }
]
keywords = ["cpl", "auth", "backend", "shared", "library"]
dynamic = ["dependencies", "optional-dependencies"]
[project.urls]
Homepage = "https://www.sh-edraft.de"
[tool.setuptools.packages.find]
where = ["."]
include = ["cpl*"]
[tool.setuptools.dynamic]
dependencies = { file = ["requirements.txt"] }
optional-dependencies.dev = { file = ["requirements.dev.txt"] }

View File

@@ -0,0 +1 @@
black==25.1.0

View File

@@ -0,0 +1,4 @@
cpl-core
cpl-dependency
cpl-database
python-keycloak-5.8.1

View File

@@ -1 +0,0 @@

View File

@@ -2,14 +2,13 @@ import inspect
import json
import os
import sys
from inspect import isclass
from typing import Any
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.core.console.console import Console
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
from cpl.core.environment.environment import Environment
from cpl.core.typing import D, T
from cpl.core.utils.json_processor import JSONProcessor
class Configuration:
@@ -88,6 +87,8 @@ class Configuration:
if os.path.isabs(name):
file_path = name
else:
from cpl.core.environment import Environment
path_root = Environment.get_cwd()
if path is not None:
path_root = path
@@ -115,9 +116,7 @@ class Configuration:
if sub.__name__ != key and sub.__name__.replace("Settings", "") != key:
continue
configuration = JSONProcessor.process(sub, value)
cls.set(sub, configuration)
cls.set(sub, sub(value))
@classmethod
def set(cls, key: Any, value: T):
@@ -128,7 +127,11 @@ class Configuration:
@classmethod
def get(cls, key: Any, default: D = None) -> T | D:
if inspect.isclass(key):
key = key.__name__
key_name = key.__name__ if inspect.isclass(key) else key
return cls._config.get(key, default)
result = cls._config.get(key_name, default)
if issubclass(key, ConfigurationModelABC) and result == default:
result = key()
cls.set(key, result)
return result

View File

@@ -1,5 +1,82 @@
from abc import ABC
from abc import ABC, abstractmethod
from typing import Optional, Type, Any
from cpl.core.typing import T
from cpl.core.utils.cast import cast
from cpl.core.utils.get_value import get_value
from cpl.core.utils.string import String
class ConfigurationModelABC(ABC):
pass
r"""
ABC for configuration model classes
"""
@abstractmethod
def __init__(
self,
src: Optional[dict] = None,
env_prefix: Optional[str] = None,
readonly: bool = True,
):
ABC.__init__(self)
self._src = src or {}
self._options: dict[str, Any] = {}
self._env_prefix = env_prefix
self._readonly = readonly
def __setattr__(self, attr: str, value: Any):
if hasattr(self, "_readonly") and self._readonly:
raise AttributeError(f"Cannot set attribute: {attr}. {type(self).__name__} is read-only")
super().__setattr__(attr, value)
def __getattr__(self, attr: str) -> Any:
options = super().__getattribute__("_options")
if attr in options:
return options[attr]
return super().__getattribute__(attr)
def option(self, field: str, cast_type: Type[T], default=None, required=False, from_env=True):
value = None
field_variants = [
field,
String.first_to_upper(field),
String.first_to_lower(field),
String.to_camel_case(field),
String.to_snake_case(field),
String.to_pascal_case(field),
]
value = None
for variant in field_variants:
if variant in self._src:
value = self._src[variant]
break
if value is None and from_env:
from cpl.core.environment import Environment
env_field = field.upper()
if self._env_prefix:
env_field = f"{self._env_prefix}_{env_field}"
value = cast(Environment.get(env_field, str), cast_type)
if value is None and required:
raise ValueError(f"{field} is required")
elif value is None:
self._options[field] = default
return
self._options[field] = cast(value, cast_type)
def get(self, field: str, default=None) -> Optional[T]:
return get_value(self._src, field, self._options[field].type, default)
def to_dict(self) -> dict:
return {field: self.get(field) for field in self._options.keys()}

View File

@@ -1,5 +1,4 @@
from .background_color_enum import BackgroundColorEnum
from .console import Console
from .console_call import ConsoleCall
from ._call import ConsoleCall
from .foreground_color_enum import ForegroundColorEnum
from .spinner_thread import SpinnerThread

View File

@@ -1,7 +1,8 @@
import os
import sys
import threading
import multiprocessing
import time
from multiprocessing import Process
from termcolor import colored
@@ -9,8 +10,8 @@ from cpl.core.console.background_color_enum import BackgroundColorEnum
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
class SpinnerThread(threading.Thread):
r"""Thread to show spinner in terminal
class Spinner(Process):
r"""Process to show spinner in terminal
Parameter:
msg_len: :class:`int`
@@ -22,7 +23,7 @@ class SpinnerThread(threading.Thread):
"""
def __init__(self, msg_len: int, foreground_color: ForegroundColorEnum, background_color: BackgroundColorEnum):
threading.Thread.__init__(self)
Process.__init__(self)
self._msg_len = msg_len
self._foreground_color = foreground_color
@@ -50,29 +51,26 @@ class SpinnerThread(threading.Thread):
return color_args
def run(self) -> None:
r"""Entry point of thread, shows the spinner"""
r"""Entry point of process, shows the spinner"""
columns = 0
if sys.platform == "win32":
columns = os.get_terminal_size().columns
else:
term_rows, term_columns = os.popen("stty size", "r").read().split()
values = os.popen("stty size", "r").read().split()
term_rows, term_columns = values if len(values) == 2 else (0, 0)
columns = int(term_columns)
end_msg = "done"
end_msg_pos = columns - self._msg_len - len(end_msg)
if end_msg_pos > 0:
print(f'{"" : >{end_msg_pos}}', end="")
padding = columns - self._msg_len - len(end_msg)
if padding > 0:
print(f'{"" : >{padding}}', end="")
else:
print("", end="")
first = True
spinner = self._spinner()
while self._is_spinning:
if first:
first = False
print(colored(f"{next(spinner): >{len(end_msg) - 1}}", *self._get_color_args()), end="")
else:
print(colored(f"{next(spinner): >{len(end_msg)}}", *self._get_color_args()), end="")
print(colored(f"{next(spinner): >{len(end_msg)}}", *self._get_color_args()), end="")
time.sleep(0.1)
back = ""
for i in range(0, len(end_msg)):
@@ -84,9 +82,10 @@ class SpinnerThread(threading.Thread):
if not self._exit:
print(colored(end_msg, *self._get_color_args()), end="")
def stop_spinning(self):
def stop(self):
r"""Stops the spinner"""
self._is_spinning = False
super().terminate()
time.sleep(0.1)
def exit(self):

View File

@@ -10,9 +10,9 @@ from tabulate import tabulate
from termcolor import colored
from cpl.core.console.background_color_enum import BackgroundColorEnum
from cpl.core.console.console_call import ConsoleCall
from cpl.core.console._call import ConsoleCall
from cpl.core.console.foreground_color_enum import ForegroundColorEnum
from cpl.core.console.spinner_thread import SpinnerThread
from cpl.core.console._spinner import Spinner
class Console:
@@ -464,7 +464,7 @@ class Console:
cls.set_hold_back(True)
spinner = None
if not cls._disabled:
spinner = SpinnerThread(len(message), spinner_foreground_color, spinner_background_color)
spinner = Spinner(len(message), spinner_foreground_color, spinner_background_color)
spinner.start()
return_value = None
@@ -476,7 +476,7 @@ class Console:
cls.close()
if spinner is not None:
spinner.stop_spinning()
spinner.stop()
cls.set_hold_back(False)
cls.set_foreground_color(ForegroundColorEnum.default)

View File

@@ -0,0 +1 @@
from .user_context import set_user, get_user

View File

@@ -0,0 +1,18 @@
from contextvars import ContextVar
from typing import Optional
from cpl.auth.auth_logger import AuthLogger
from cpl.auth.schema._administration.auth_user import AuthUser
_user_context: ContextVar[Optional[AuthUser]] = ContextVar("user", default=None)
_logger = AuthLogger(__name__)
def set_user(user_id: Optional[AuthUser]):
_logger.trace("Setting user context", user_id)
_user_context.set(user_id)
def get_user() -> Optional[AuthUser]:
return _user_context.get()

View File

@@ -1,9 +1,9 @@
import os
from socket import gethostname
from typing import Optional, Type
from typing import Type
from cpl.core.environment.environment_enum import EnvironmentEnum
from cpl.core.typing import T
from cpl.core.typing import T, D
from cpl.core.utils.get_value import get_value
@@ -55,14 +55,14 @@ class Environment:
os.environ[key] = str(value)
@staticmethod
def get(key: str, cast_type: Type[T], default: Optional[T] = None) -> Optional[T]:
def get(key: str, cast_type: Type[T], default: D = None) -> T | D:
"""
Get an environment variable and cast it to a specified type.
:param str key: The name of the environment variable.
:param Type[T] cast_type: A callable to cast the variable's value.
:param Optional[T] default: The default value to return if the variable is not found. Defaults to None.The default value to return if the variable is not found. Defaults to None.
:param T default: The default value to return if the variable is not found. Defaults to None.The default value to return if the variable is not found. Defaults to None.
:return: The casted value, or None if the variable is not found.
:rtype: Optional[T]
:rtype: T | D
"""
return get_value(dict(os.environ), key, cast_type, default)

View File

@@ -1,4 +1,4 @@
from .logger import Logger
from .logger_abc import LoggerABC
from .log_level_enum import LogLevelEnum
from .logging_settings import LogSettings
from .log_level import LogLevel
from .log_settings import LogSettings

View File

@@ -1,92 +0,0 @@
import multiprocessing
import os
from datetime import datetime
from typing import Self
from cpl.core.console import Console
from cpl.core.log.log_level_enum import LogLevelEnum
class LogWriter:
_instance = None
# ANSI color codes for different log levels
_COLORS = {
LogLevelEnum.trace: "\033[37m", # Light Gray
LogLevelEnum.debug: "\033[94m", # Blue
LogLevelEnum.info: "\033[92m", # Green
LogLevelEnum.warning: "\033[93m", # Yellow
LogLevelEnum.error: "\033[91m", # Red
LogLevelEnum.fatal: "\033[95m", # Magenta
}
def __init__(self, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info):
self._file_prefix = file_prefix
self._level = level
self._queue = multiprocessing.Queue()
self._process = multiprocessing.Process(target=self._log_worker, daemon=True)
self._create_log_dir()
self._process.start()
@property
def level(self) -> LogLevelEnum:
return self._level
@level.setter
def level(self, value: LogLevelEnum):
assert isinstance(value, LogLevelEnum), "Log level must be an instance of LogLevelEnum"
self._level = value
@classmethod
def get_instance(cls, file_prefix: str, level: LogLevelEnum = LogLevelEnum.info) -> Self:
if cls._instance is None:
cls._instance = LogWriter(file_prefix, level)
return cls._instance
@staticmethod
def _create_log_dir():
if os.path.exists("logs"):
return
os.makedirs("logs")
def _log_worker(self):
"""Worker process that writes log messages from the queue to the file."""
while True:
content = self._queue.get()
if content is None: # Shutdown signal
break
self._write_log_to_file(content)
Console.write_line(f"{self._COLORS.get(self._level, '\033[0m')}{content}\033[0m")
@property
def log_file(self):
return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
def _ensure_file_size(self):
log_file = self.log_file
if not os.path.exists(log_file) or os.path.getsize(log_file) <= 0.5 * 1024 * 1024:
return
# if exists and size is greater than 300MB, create a new file
os.rename(
log_file,
f"{log_file.split('.log')[0]}_{datetime.now().strftime('%H-%M-%S')}.log",
)
def _write_log_to_file(self, content: str):
self._ensure_file_size()
with open(self.log_file, "a") as log_file:
log_file.write(content + "\n")
log_file.close()
def log(self, content: str):
"""Enqueue log message without blocking main app."""
self._queue.put(content)
def close(self):
"""Gracefully stop the logging process."""
self._queue.put(None)
self._process.join()

View File

@@ -1,7 +1,7 @@
from enum import Enum
class LogLevelEnum(Enum):
class LogLevel(Enum):
off = "OFF" # Nothing
trace = "TRC" # Detailed app information's
debug = "DEB" # Detailed app state

View File

@@ -0,0 +1,18 @@
from typing import Optional
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.core.log.log_level import LogLevel
class LogSettings(ConfigurationModelABC):
def __init__(
self,
src: Optional[dict] = None,
):
ConfigurationModelABC.__init__(self, src, "LOG")
self.option("path", str, default="logs")
self.option("filename", str, default="app.log")
self.option("console", LogLevel, default=LogLevel.info)
self.option("level", LogLevel, default=LogLevel.info)

View File

@@ -3,43 +3,100 @@ import traceback
from datetime import datetime
from cpl.core.console import Console
from cpl.core.log._log_writer import LogWriter
from cpl.core.log.log_level_enum import LogLevelEnum
from cpl.core.log.log_level import LogLevel
from cpl.core.log.logger_abc import LoggerABC
from cpl.core.typing import Messages, Source
class Logger(LoggerABC):
_level = LogLevelEnum.info
_levels = [x for x in LogLevelEnum]
_levels = [x for x in LogLevel]
# ANSI color codes for different log levels
_COLORS = {
LogLevel.trace: "\033[37m", # Light Gray
LogLevel.debug: "\033[94m", # Blue
LogLevel.info: "\033[92m", # Green
LogLevel.warning: "\033[93m", # Yellow
LogLevel.error: "\033[91m", # Red
LogLevel.fatal: "\033[95m", # Magenta
}
def __init__(self, source: Source, file_prefix: str = None):
LoggerABC.__init__(self)
assert source is not None and source != "", "Source cannot be None or empty"
if source == LoggerABC.__name__:
source = None
self._source = source
if file_prefix is None:
file_prefix = "app"
self._file_prefix = file_prefix
self._writer = LogWriter.get_instance(self._file_prefix)
self._create_log_dir()
@property
def _settings(self):
from cpl.core.configuration.configuration import Configuration
from cpl.core.log.log_settings import LogSettings
return Configuration.get(LogSettings)
@property
def log_file(self):
return f"logs/{self._file_prefix}_{datetime.now().strftime('%Y-%m-%d')}.log"
@staticmethod
def _create_log_dir():
if os.path.exists("logs"):
return
os.makedirs("logs")
@classmethod
def set_level(cls, level: LogLevelEnum):
def set_level(cls, level: LogLevel):
if level in cls._levels:
cls._level = level
else:
raise ValueError(f"Invalid log level: {level}")
def _log(self, level: LogLevelEnum, *messages: Messages):
try:
if self._levels.index(level) < self._levels.index(self._level):
return
@staticmethod
def _ensure_file_size(log_file: str):
if not os.path.exists(log_file) or os.path.getsize(log_file) <= 0.5 * 1024 * 1024:
return
# if exists and size is greater than 300MB, create a new file
os.rename(
log_file,
f"{log_file.split('.log')[0]}_{datetime.now().strftime('%H-%M-%S')}.log",
)
def _should_log(self, input_level: LogLevel, settings_level: LogLevel) -> bool:
return self._levels.index(input_level) >= self._levels.index(settings_level)
def _write_log_to_file(self, level: LogLevel, content: str):
if not self._should_log(level, self._settings.level):
return
file = self.log_file
self._ensure_file_size(file)
with open(file, "a") as log_file:
log_file.write(content + "\n")
log_file.close()
def _write_to_console(self, level: LogLevel, content: str):
if not self._should_log(level, self._settings.console):
return
Console.write_line(f"{self._COLORS.get(level, '\033[0m')}{content}\033[0m")
def _log(self, level: LogLevel, *messages: Messages):
try:
timestamp = datetime.now().strftime("%Y-%m-%d %H:%M:%S.%f")
formatted_message = self._format_message(level.value, timestamp, *messages)
self._writer.log(formatted_message)
self._write_log_to_file(level, formatted_message)
self._write_to_console(level, formatted_message)
except Exception as e:
print(f"Error while logging: {e} -> {traceback.format_exc()}")
@@ -52,27 +109,35 @@ class Logger(LoggerABC):
messages = [str(message) for message in messages if message is not None]
return f"<{timestamp}> [{level.upper():^3}] [{self._file_prefix}] - [{self._source}]: {' '.join(messages)}"
message = f"<{timestamp}>"
message += f" [{level.upper():^3}]"
message += f" [{self._file_prefix}]"
if self._source is not None:
message += f" - [{self._source}]"
message += f": {' '.join(messages)}"
return message
def header(self, string: str):
self._log(LogLevelEnum.info, string)
self._log(LogLevel.info, string)
def trace(self, *messages: Messages):
self._log(LogLevelEnum.trace, *messages)
self._log(LogLevel.trace, *messages)
def debug(self, *messages: Messages):
self._log(LogLevelEnum.debug, *messages)
self._log(LogLevel.debug, *messages)
def info(self, *messages: Messages):
self._log(LogLevelEnum.info, *messages)
self._log(LogLevel.info, *messages)
def warning(self, *messages: Messages):
self._log(LogLevelEnum.warning, *messages)
self._log(LogLevel.warning, *messages)
def error(self, message, e: Exception = None):
self._log(LogLevelEnum.error, message, f"{e} -> {traceback.format_exc()}" if e else None)
self._log(LogLevel.error, message, f"{e} -> {traceback.format_exc()}" if e else None)
def fatal(self, message, e: Exception = None, prevent_quit: bool = False):
self._log(LogLevelEnum.fatal, message, f"{e} -> {traceback.format_exc()}" if e else None)
self._log(LogLevel.fatal, message, f"{e} -> {traceback.format_exc()}" if e else None)
if not prevent_quit:
exit(-1)

View File

@@ -1,5 +1,6 @@
from abc import abstractmethod, ABC
from cpl.core.log.log_level import LogLevel
from cpl.core.typing import Messages
@@ -7,12 +8,10 @@ class LoggerABC(ABC):
r"""ABC for :class:`cpl.core.log.logger_service.Logger`"""
@abstractmethod
def set_level(self, level: str):
pass
def set_level(self, level: LogLevel): ...
@abstractmethod
def _format_message(self, level: str, timestamp, *messages: Messages) -> str:
pass
def _format_message(self, level: str, timestamp, *messages: Messages) -> str: ...
@abstractmethod
def header(self, string: str):

View File

@@ -1,53 +0,0 @@
from typing import Optional
from cpl.core.configuration.configuration_model_abc import ConfigurationModelABC
from cpl.core.log.log_level_enum import LogLevelEnum
class LogSettings(ConfigurationModelABC):
r"""Representation of logging settings"""
def __init__(
self,
path: str = None,
filename: str = None,
console_log_level: LogLevelEnum = None,
file_log_level: LogLevelEnum = None,
):
ConfigurationModelABC.__init__(self)
self._path: Optional[str] = path
self._filename: Optional[str] = filename
self._console: Optional[LogLevelEnum] = console_log_level
self._level: Optional[LogLevelEnum] = file_log_level
@property
def path(self) -> str:
return self._path
@path.setter
def path(self, path: str) -> None:
self._path = path
@property
def filename(self) -> str:
return self._filename
@filename.setter
def filename(self, filename: str) -> None:
self._filename = filename
@property
def console(self) -> LogLevelEnum:
return self._console
@console.setter
def console(self, console: LogLevelEnum) -> None:
self._console = console
@property
def level(self) -> LogLevelEnum:
return self._level
@level.setter
def level(self, level: LogLevelEnum) -> None:
self._level = level

View File

@@ -7,10 +7,8 @@ from cpl.core.typing import T
class PipeABC(ABC, Generic[T]):
@staticmethod
@abstractmethod
def to_str(value: T, *args) -> str:
pass
def to_str(value: T, *args) -> str: ...
@staticmethod
@abstractmethod
def from_str(value: str, *args) -> T:
pass
def from_str(value: str, *args) -> T: ...

View File

@@ -1,4 +1,5 @@
from typing import TypeVar, Any
from uuid import UUID
T = TypeVar("T")
D = TypeVar("D")
@@ -8,3 +9,8 @@ Service = TypeVar("Service")
Source = TypeVar("Source")
Messages = list[Any] | Any
UuidId = str | UUID
SerialId = int
Id = UuidId | SerialId

View File

@@ -1,5 +1,6 @@
from .b64 import B64
from .base64 import Base64
from .credential_manager import CredentialManager
from .json_processor import JSONProcessor
from .pip import Pip
from .string import String
from .get_value import get_value

View File

@@ -2,7 +2,7 @@ import base64
from typing import Union
class B64:
class Base64:
@staticmethod
def encode(string: str) -> str:

View File

@@ -0,0 +1,69 @@
from enum import Enum
from typing import Type, Any
from cpl.core.typing import T
def _cast_enum(value: str, enum_type: Type[Enum]) -> Enum:
try:
return enum_type(value)
except ValueError:
pass
try:
return enum_type(value.lower())
except ValueError:
pass
try:
return enum_type(value.upper())
except ValueError:
pass
try:
return enum_type[value]
except KeyError:
pass
try:
return enum_type[value.lower()]
except KeyError:
pass
try:
return enum_type[value.upper()]
except KeyError:
pass
raise ValueError(f"Cannot cast value '{value}' to enum '{enum_type.__name__}'")
def cast(value: Any, cast_type: Type[T], list_delimiter: str = ",") -> T:
"""
Cast a value to a specified type.
:param Any value: Value to be casted.
:param Type[T] cast_type: A callable to cast the variable's value.
:param str list_delimiter: The delimiter to split the value into a list. Defaults to ",".
:return:
"""
if value is None:
return None
if cast_type == bool:
return value.lower() in ["true", "1", "yes", "on"]
if (cast_type if not hasattr(cast_type, "__origin__") else cast_type.__origin__) == list:
if not (value.startswith("[") and value.endswith("]")) and list_delimiter not in value:
raise ValueError("List values must be enclosed in square brackets or use a delimiter.")
if value.startswith("[") and value.endswith("]"):
value = value[1:-1]
value = value.split(list_delimiter)
subtype = cast_type.__args__[0] if hasattr(cast_type, "__args__") else None
return [subtype(item) if subtype is not None else item for item in value]
if isinstance(cast_type, type) and issubclass(cast_type, Enum):
return _cast_enum(value, cast_type)
return cast_type(value)

View File

@@ -1,12 +1,42 @@
import base64
import os
from cryptography.fernet import Fernet
from cpl.core.log.logger import Logger
_logger = Logger(__name__)
class CredentialManager:
r"""Handles credential encryption and decryption"""
@staticmethod
def encrypt(string: str) -> str:
r"""Encode with base64
_secret: str = None
@classmethod
def with_secret(cls, file: str = None):
if file is None:
file = ".secret"
if not os.path.isfile(file):
dirname = os.path.dirname(file)
if dirname != "":
os.makedirs(dirname, exist_ok=True)
with open(file, "w") as secret_file:
secret_file.write(Fernet.generate_key().decode())
secret_file.close()
_logger.warning("Secret file not found, regenerating")
with open(file, "r") as secret_file:
secret = secret_file.read().strip()
if secret == "" or secret is None:
_logger.fatal("No secret found in .secret file.")
cls._secret = str(secret)
@classmethod
def encrypt(cls, string: str) -> str:
r"""Encode with Fernet
Parameter:
string: :class:`str`
@@ -15,11 +45,11 @@ class CredentialManager:
Returns:
Encoded string
"""
return base64.b64encode(string.encode("utf-8")).decode("utf-8")
return Fernet(cls._secret).encrypt(string.encode()).decode()
@staticmethod
def decrypt(string: str) -> str:
r"""Decode with base64
@classmethod
def decrypt(cls, string: str) -> str:
r"""Decode with Fernet
Parameter:
string: :class:`str`
@@ -28,19 +58,4 @@ class CredentialManager:
Returns:
Decoded string
"""
return base64.b64decode(string).decode("utf-8")
@staticmethod
def build_string(string: str, credentials: str):
r"""Builds string with credentials in it
Parameter:
string: :class:`str`
String in which the variable is replaced by credentials
credentials: :class:`str`
String to encode
Returns:
Decoded string
"""
return string.replace("$credentials", CredentialManager.decrypt(credentials))
return Fernet(cls._secret).decrypt(string).decode()

View File

@@ -1,6 +1,7 @@
from typing import Type, Optional
from cpl.core.typing import T
from cpl.core.utils.cast import cast
def get_value(
@@ -37,20 +38,9 @@ def get_value(
return value
try:
if cast_type == bool:
return value.lower() in ["true", "1"]
if (cast_type if not hasattr(cast_type, "__origin__") else cast_type.__origin__) == list:
if not (value.startswith("[") and value.endswith("]")) and list_delimiter not in value:
raise ValueError("List values must be enclosed in square brackets or use a delimiter.")
if value.startswith("[") and value.endswith("]"):
value = value[1:-1]
value = value.split(list_delimiter)
subtype = cast_type.__args__[0] if hasattr(cast_type, "__args__") else None
return [subtype(item) if subtype is not None else item for item in value]
return cast_type(value)
cast(value, cast_type, list_delimiter)
except (ValueError, TypeError):
from cpl.core.log import Logger
Logger(__name__).debug(f"Failed to cast value '{value}' to type '{cast_type.__name__}'")
return default

View File

@@ -17,7 +17,36 @@ class String:
Returns:
String converted to CamelCase
"""
return re.sub(r"(?<!^)(?=[A-Z])", "_", s).lower()
parts = re.split(r"[^a-zA-Z0-9]+", s.strip())
parts = [p for p in parts if p]
if not parts:
return ""
return parts[0].lower() + "".join(word.capitalize() for word in parts[1:])
@staticmethod
def to_pascal_case(s: str) -> str:
r"""Converts string to pascal case
Parameter:
chars: :class:`str`
String to convert
Returns:
String converted to PascalCase
"""
parts = re.split(r"[^a-zA-Z0-9]+", s.strip())
parts = [p for p in parts if p]
if not parts:
return ""
return "".join(word.capitalize() for word in parts)
@staticmethod
def to_snake_case(chars: str) -> str:

View File

@@ -1,3 +1,70 @@
from .database_settings_name_enum import DatabaseSettingsNameEnum
from .database_settings import DatabaseSettings
from .table_abc import TableABC
from typing import Type
from cpl.application.abc import ApplicationABC as _ApplicationABC
from cpl.dependency import ServiceCollection as _ServiceCollection
from . import mysql as _mysql
from . import postgres as _postgres
from .table_manager import TableManager
def _with_migrations(self: _ApplicationABC, *paths: list[str]) -> _ApplicationABC:
from cpl.application.host import Host
from cpl.database.service.migration_service import MigrationService
migration_service = self._services.get_service(MigrationService)
migration_service.with_directory("./scripts")
Host.run(migration_service.migrate)
return self
def _with_seeders(self: _ApplicationABC) -> _ApplicationABC:
from cpl.database.service.seeder_service import SeederService
from cpl.application.host import Host
seeder_service: SeederService = self._services.get_service(SeederService)
Host.run(seeder_service.seed)
return self
def _add(collection: _ServiceCollection, db_context: Type, default_port: int, server_type: str):
from cpl.core.console import Console
from cpl.core.configuration import Configuration
from cpl.database.abc.db_context_abc import DBContextABC
from cpl.database.model.server_type import ServerTypes, ServerType
from cpl.database.model.database_settings import DatabaseSettings
from cpl.database.service.migration_service import MigrationService
from cpl.database.service.seeder_service import SeederService
from cpl.database.schema.executed_migration_dao import ExecutedMigrationDao
try:
ServerType.set_server_type(ServerTypes(server_type))
Configuration.set("DB_DEFAULT_PORT", default_port)
collection.add_singleton(DBContextABC, db_context)
collection.add_singleton(ExecutedMigrationDao)
collection.add_singleton(MigrationService)
collection.add_singleton(SeederService)
except ImportError as e:
Console.error("cpl-database is not installed", str(e))
def add_mysql(collection: _ServiceCollection):
from cpl.database.mysql.db_context import DBContext
from cpl.database.model import ServerTypes
_add(collection, DBContext, 3306, ServerTypes.MYSQL.value)
def add_postgres(collection: _ServiceCollection):
from cpl.database.mysql.db_context import DBContext
from cpl.database.model import ServerTypes
_add(collection, DBContext, 5432, ServerTypes.POSTGRES.value)
_ServiceCollection.with_module(add_mysql, _mysql.__name__)
_ServiceCollection.with_module(add_postgres, _postgres.__name__)
_ApplicationABC.extend(_ApplicationABC.with_migrations, _with_migrations)
_ApplicationABC.extend(_ApplicationABC.with_seeders, _with_seeders)

View File

@@ -0,0 +1,5 @@
from .connection_abc import ConnectionABC
from .db_context_abc import DBContextABC
from .db_join_model_abc import DbJoinModelABC
from .db_model_abc import DbModelABC
from .db_model_dao_abc import DbModelDaoABC

View File

@@ -1,26 +1,23 @@
from abc import ABC, abstractmethod
from cpl.database.database_settings import DatabaseSettings
from cpl.database.model.database_settings import DatabaseSettings
from mysql.connector.abstracts import MySQLConnectionAbstract
from mysql.connector.cursor import MySQLCursorBuffered
class DatabaseConnectionABC(ABC):
class ConnectionABC(ABC):
r"""ABC for the :class:`cpl.database.connection.database_connection.DatabaseConnection`"""
@abstractmethod
def __init__(self):
pass
def __init__(self): ...
@property
@abstractmethod
def server(self) -> MySQLConnectionAbstract:
pass
def server(self) -> MySQLConnectionAbstract: ...
@property
@abstractmethod
def cursor(self) -> MySQLCursorBuffered:
pass
def cursor(self) -> MySQLCursorBuffered: ...
@abstractmethod
def connect(self, database_settings: DatabaseSettings):

View File

@@ -0,0 +1,877 @@
import datetime
from abc import ABC, abstractmethod
from enum import Enum
from types import NoneType
from typing import Generic, Optional, Union, Type, List, Any
from cpl.core.typing import T, Id
from cpl.core.utils.get_value import get_value
from cpl.core.utils.string import String
from cpl.database.abc.db_context_abc import DBContextABC
from cpl.database.const import DATETIME_FORMAT
from cpl.database.db_logger import DBLogger
from cpl.database.external_data_temp_table_builder import ExternalDataTempTableBuilder
from cpl.database.postgres.sql_select_builder import SQLSelectBuilder
from cpl.database.typing import T_DBM, Attribute, AttributeFilters, AttributeSorts
class DataAccessObjectABC(ABC, Generic[T_DBM]):
@abstractmethod
def __init__(self, source: str, model_type: Type[T_DBM], table_name: str):
from cpl.dependency.service_provider_abc import ServiceProviderABC
self._db = ServiceProviderABC.get_global_service(DBContextABC)
self._logger = DBLogger(source)
self._model_type = model_type
self._table_name = table_name
self._logger = DBLogger(source)
self._model_type = model_type
self._table_name = table_name
self._default_filter_condition = None
self.__attributes: dict[str, type] = {}
self.__db_names: dict[str, str] = {}
self.__foreign_tables: dict[str, tuple[str, str]] = {}
self.__foreign_table_keys: dict[str, str] = {}
self.__foreign_dao: dict[str, "DataAccessObjectABC"] = {}
self.__date_attributes: set[str] = set()
self.__ignored_attributes: set[str] = set()
self.__primary_key = "id"
self.__primary_key_type = int
self._external_fields: dict[str, ExternalDataTempTableBuilder] = {}
@property
def table_name(self) -> str:
return self._table_name
def has_attribute(self, attr_name: Attribute) -> bool:
"""
Check if the attribute exists in the DAO
:param Attribute attr_name: Name of the attribute
:return: True if the attribute exists, False otherwise
"""
return attr_name in self.__attributes
def attribute(
self,
attr_name: Attribute,
attr_type: type,
db_name: str = None,
ignore=False,
primary_key=False,
aliases: list[str] = None,
):
"""
Add an attribute for db and object mapping to the data access object
:param Attribute attr_name: Name of the attribute in the object
:param type attr_type: Python type of the attribute to cast db value to
:param str db_name: Name of the field in the database, if None the attribute lowered attr_name without "_" is used
:param bool ignore: Defines if field is ignored for create and update (for e.g. auto increment fields or created/updated fields)
:param bool primary_key: Defines if field is the primary key
:param list[str] aliases: List of aliases for the attribute name
:return:
"""
if isinstance(attr_name, property):
attr_name = attr_name.fget.__name__
self.__attributes[attr_name] = attr_type
if ignore:
self.__ignored_attributes.add(attr_name)
if not db_name:
db_name = attr_name.lower().replace("_", "")
self.__db_names[attr_name] = db_name
self.__db_names[db_name] = db_name
if aliases is not None:
for alias in aliases:
if alias in self.__db_names:
raise ValueError(f"Alias {alias} already exists")
self.__db_names[alias] = db_name
if primary_key:
self.__primary_key = db_name
self.__primary_key_type = attr_type
if attr_type in [datetime, datetime.datetime]:
self.__date_attributes.add(attr_name)
self.__date_attributes.add(db_name)
def reference(
self,
attr: Attribute,
primary_attr: Attribute,
foreign_attr: Attribute,
table_name: str,
reference_dao: "DataAccessObjectABC" = None,
):
"""
Add a reference to another table for the given attribute
:param Attribute attr: Name of the attribute in the object
:param str primary_attr: Name of the primary key in the foreign object
:param str foreign_attr: Name of the foreign key in the object
:param str table_name: Name of the table to reference
:param DataAccessObjectABC reference_dao: The data access object for the referenced table
:return:
"""
if isinstance(attr, property):
attr = attr.fget.__name__
if isinstance(primary_attr, property):
primary_attr = primary_attr.fget.__name__
primary_attr = primary_attr.lower().replace("_", "")
if isinstance(foreign_attr, property):
foreign_attr = foreign_attr.fget.__name__
foreign_attr = foreign_attr.lower().replace("_", "")
self.__foreign_table_keys[attr] = foreign_attr
if reference_dao is not None:
self.__foreign_dao[attr] = reference_dao
if table_name == self._table_name:
return
self.__foreign_tables[attr] = (
table_name,
f"{table_name}.{primary_attr} = {self._table_name}.{foreign_attr}",
)
def use_external_fields(self, builder: ExternalDataTempTableBuilder):
self._external_fields[builder.table_name] = builder
def to_object(self, result: dict) -> T_DBM:
"""
Convert a result from the database to an object
:param dict result: Result from the database
:return:
"""
value_map: dict[str, T] = {}
for db_name, value in result.items():
# Find the attribute name corresponding to the db_name
attr_name = next((k for k, v in self.__db_names.items() if v == db_name), None)
if attr_name:
value_map[attr_name] = self._get_value_from_sql(self.__attributes[attr_name], value)
return self._model_type(**value_map)
def to_dict(self, obj: T_DBM) -> dict:
"""
Convert an object to a dictionary
:param T_DBM obj: Object to convert
:return:
"""
value_map: dict[str, Any] = {}
for attr_name, attr_type in self.__attributes.items():
value = getattr(obj, attr_name)
if isinstance(value, datetime.datetime):
value = value.strftime(DATETIME_FORMAT)
elif isinstance(value, Enum):
value = value.value
value_map[attr_name] = value
for ex_fname in self._external_fields:
ex_field = self._external_fields[ex_fname]
for ex_attr in ex_field.fields:
if ex_attr == self.__primary_key:
continue
value_map[ex_attr] = getattr(obj, ex_attr, None)
return value_map
async def count(self, filters: AttributeFilters = None) -> int:
result = await self._prepare_query(filters=filters, for_count=True)
return result[0]["count"] if result else 0
async def get_history(
self,
entry_id: int,
by_key: str = None,
when: datetime = None,
until: datetime = None,
without_deleted: bool = False,
) -> list[T_DBM]:
"""
Retrieve the history of an entry from the history table.
:param entry_id: The ID of the entry to retrieve history for.
:param by_key: The key to filter by (default is the primary key).
:param when: A specific timestamp to filter the history.
:param until: A timestamp to filter history entries up to a certain point.
:param without_deleted: Exclude deleted entries if True.
:return: A list of historical entries as objects.
"""
f_tables = list(self.__foreign_tables.keys())
history_table = f"{self._table_name}_history"
builder = SQLSelectBuilder(history_table, self.__primary_key)
builder.with_attribute("*")
builder.with_value_condition(
f"{history_table}.{by_key or self.__primary_key}",
"=",
str(entry_id),
f_tables,
)
if self._default_filter_condition:
builder.with_condition(self._default_filter_condition, "", f_tables)
if without_deleted:
builder.with_value_condition(f"{history_table}.deleted", "=", "false", f_tables)
if when:
builder.with_value_condition(
self._attr_from_date_to_char(f"{history_table}.updated"),
"=",
f"'{when.strftime(DATETIME_FORMAT)}'",
f_tables,
)
if until:
builder.with_value_condition(
self._attr_from_date_to_char(f"{history_table}.updated"),
"<=",
f"'{until.strftime(DATETIME_FORMAT)}'",
f_tables,
)
builder.with_order_by(f"{history_table}.updated", "DESC")
query = await builder.build()
result = await self._db.select_map(query)
return [self.to_object(x) for x in result] if result else []
async def get_all(self) -> List[T_DBM]:
result = await self._prepare_query(sorts=[{self.__primary_key: "asc"}])
return [self.to_object(x) for x in result] if result else []
async def get_by_id(self, id: Union[int, str]) -> Optional[T_DBM]:
result = await self._prepare_query(filters=[{self.__primary_key: id}], sorts=[{self.__primary_key: "asc"}])
return self.to_object(result[0]) if result else None
async def find_by_id(self, id: Union[int, str]) -> Optional[T_DBM]:
result = await self._prepare_query(filters=[{self.__primary_key: id}], sorts=[{self.__primary_key: "asc"}])
return self.to_object(result[0]) if result else None
async def get_by(
self,
filters: AttributeFilters = None,
sorts: AttributeSorts = None,
take: int = None,
skip: int = None,
) -> list[T_DBM]:
result = await self._prepare_query(filters, sorts, take, skip)
if not result or len(result) == 0:
raise ValueError("No result found")
return [self.to_object(x) for x in result] if result else []
async def get_single_by(
self,
filters: AttributeFilters = None,
sorts: AttributeSorts = None,
take: int = None,
skip: int = None,
) -> T_DBM:
result = await self._prepare_query(filters, sorts, take, skip)
if not result:
raise ValueError("No result found")
if len(result) > 1:
raise ValueError("More than one result found")
return self.to_object(result[0])
async def find_by(
self,
filters: AttributeFilters = None,
sorts: AttributeSorts = None,
take: int = None,
skip: int = None,
) -> list[T_DBM]:
result = await self._prepare_query(filters, sorts, take, skip)
return [self.to_object(x) for x in result] if result else []
async def find_single_by(
self,
filters: AttributeFilters = None,
sorts: AttributeSorts = None,
take: int = None,
skip: int = None,
) -> Optional[T_DBM]:
result = await self._prepare_query(filters, sorts, take, skip)
if len(result) > 1:
raise ValueError("More than one result found")
return self.to_object(result[0]) if result else None
async def touch(self, obj: T_DBM):
"""
Touch the entry to update the last updated date
:return:
"""
await self._db.execute(
f"""
UPDATE {self._table_name}
SET updated = NOW()
WHERE {self.__primary_key} = {self._get_primary_key_value_sql(obj)};
"""
)
async def touch_many_by_id(self, ids: list[Id]):
"""
Touch the entries to update the last updated date
:return:
"""
if len(ids) == 0:
return
await self._db.execute(
f"""
UPDATE {self._table_name}
SET updated = NOW()
WHERE {self.__primary_key} IN ({", ".join([str(x) for x in ids])});
"""
)
async def _build_create_statement(self, obj: T_DBM, skip_editor=False) -> str:
allowed_fields = [x for x in self.__attributes.keys() if x not in self.__ignored_attributes]
fields = ", ".join([self.__db_names[x] for x in allowed_fields])
fields = f"{'EditorId' if not skip_editor else ''}{f', {fields}' if not skip_editor and len(fields) > 0 else f'{fields}'}"
values = ", ".join([self._get_value_sql(getattr(obj, x)) for x in allowed_fields])
values = f"{await self._get_editor_id(obj) if not skip_editor else ''}{f', {values}' if not skip_editor and len(values) > 0 else f'{values}'}"
return f"""
INSERT INTO {self._table_name} (
{fields}
) VALUES (
{values}
)
RETURNING {self.__primary_key};
"""
async def create(self, obj: T_DBM, skip_editor=False) -> int:
self._logger.debug(f"create {type(obj).__name__} {obj.__dict__}")
result = await self._db.execute(await self._build_create_statement(obj, skip_editor))
return self._get_value_from_sql(self.__primary_key_type, result[0][0])
async def create_many(self, objs: list[T_DBM], skip_editor=False) -> list[int]:
if len(objs) == 0:
return []
self._logger.debug(f"create many {type(objs[0]).__name__} {len(objs)} {[x.__dict__ for x in objs]}")
query = ""
for obj in objs:
query += await self._build_create_statement(obj, skip_editor)
result = await self._db.execute(query)
return [self._get_value_from_sql(self.__primary_key_type, x[0]) for x in result]
async def _build_update_statement(self, obj: T_DBM, skip_editor=False) -> str:
allowed_fields = [x for x in self.__attributes.keys() if x not in self.__ignored_attributes]
fields = ", ".join(
[f"{self.__db_names[x]} = {self._get_value_sql(getattr(obj, x, None))}" for x in allowed_fields]
)
fields = f"{f'EditorId = {await self._get_editor_id(obj)}' if not skip_editor else ''}{f', {fields}' if not skip_editor and len(fields) > 0 else f'{fields}'}"
return f"""
UPDATE {self._table_name}
SET {fields}
WHERE {self.__primary_key} = {self._get_primary_key_value_sql(obj)};
"""
async def update(self, obj: T_DBM, skip_editor=False):
self._logger.debug(f"update {type(obj).__name__} {obj.__dict__}")
await self._db.execute(await self._build_update_statement(obj, skip_editor))
async def update_many(self, objs: list[T_DBM], skip_editor=False):
if len(objs) == 0:
return
self._logger.debug(f"update many {type(objs[0]).__name__} {len(objs)} {[x.__dict__ for x in objs]}")
query = ""
for obj in objs:
query += await self._build_update_statement(obj, skip_editor)
await self._db.execute(query)
async def _build_delete_statement(self, obj: T_DBM, hard_delete: bool = False) -> str:
if hard_delete:
return f"""
DELETE FROM {self._table_name}
WHERE {self.__primary_key} = {self._get_primary_key_value_sql(obj)};
"""
return f"""
UPDATE {self._table_name}
SET EditorId = {await self._get_editor_id(obj)},
Deleted = true
WHERE {self.__primary_key} = {self._get_primary_key_value_sql(obj)};
"""
async def delete(self, obj: T_DBM, hard_delete: bool = False):
self._logger.debug(f"delete {type(obj).__name__} {obj.__dict__}")
await self._db.execute(await self._build_delete_statement(obj, hard_delete))
async def delete_many(self, objs: list[T_DBM], hard_delete: bool = False):
if len(objs) == 0:
return
self._logger.debug(f"delete many {type(objs[0]).__name__} {len(objs)} {[x.__dict__ for x in objs]}")
query = ""
for obj in objs:
query += await self._build_delete_statement(obj, hard_delete)
await self._db.execute(query)
async def _build_restore_statement(self, obj: T_DBM) -> str:
return f"""
UPDATE {self._table_name}
SET EditorId = {await self._get_editor_id(obj)},
Deleted = false
WHERE {self.__primary_key} = {self._get_primary_key_value_sql(obj)};
"""
async def restore(self, obj: T_DBM):
self._logger.debug(f"restore {type(obj).__name__} {obj.__dict__}")
await self._db.execute(await self._build_restore_statement(obj))
async def restore_many(self, objs: list[T_DBM]):
if len(objs) == 0:
return
self._logger.debug(f"restore many {type(objs[0]).__name__} {len(objs)} {objs[0].__dict__}")
query = ""
for obj in objs:
query += await self._build_restore_statement(obj)
await self._db.execute(query)
async def _prepare_query(
self,
filters: AttributeFilters = None,
sorts: AttributeSorts = None,
take: int = None,
skip: int = None,
for_count=False,
) -> list[dict]:
"""
Prepares and executes a query using the SQLBuilder with the given parameters.
:param filters: Conditions to filter the query.
:param sorts: Sorting attributes and directions.
:param take: Limit the number of results.
:param skip: Offset the results.
:return: Query result as a list of dictionaries.
"""
external_table_deps = []
builder = SQLSelectBuilder(self._table_name, self.__primary_key)
for temp in self._external_fields:
builder.with_temp_table(self._external_fields[temp])
if for_count:
builder.with_attribute("COUNT(*)", ignore_table_name=True)
else:
builder.with_attribute("*")
for attr in self.__foreign_tables:
table, join_condition = self.__foreign_tables[attr]
builder.with_left_join(table, join_condition)
if filters:
await self._build_conditions(builder, filters, external_table_deps)
if sorts:
self._build_sorts(builder, sorts, external_table_deps)
if take:
builder.with_limit(take)
if skip:
builder.with_offset(skip)
for external_table in external_table_deps:
builder.use_temp_table(external_table)
query = await builder.build()
return await self._db.select_map(query)
async def _build_conditions(
self,
builder: SQLSelectBuilder,
filters: AttributeFilters,
external_table_deps: list[str],
):
"""
Builds SQL conditions from GraphQL-like filters and adds them to the SQLBuilder.
:param builder: The SQLBuilder instance to add conditions to.
:param filters: GraphQL-like filter structure.
:param external_table_deps: List to store external table dependencies.
"""
if not isinstance(filters, list):
filters = [filters]
for filter_group in filters:
sql_conditions = self._graphql_to_sql_conditions(filter_group, external_table_deps)
for attr, operator, value in sql_conditions:
if attr in self.__foreign_table_keys:
attr = self.__foreign_table_keys[attr]
recursive_join = self._get_recursive_reference_join(attr)
if recursive_join is not None:
builder.with_left_join(*recursive_join)
external_table = self._get_external_field_key(attr)
if external_table is not None:
external_table_deps.append(external_table)
if operator == "fuzzy":
builder.with_levenshtein_condition(attr)
elif operator in [
"IS NULL",
"IS NOT NULL",
]: # operator without value
builder.with_condition(
attr,
operator,
[
x[0]
for fdao in self.__foreign_dao
for x in self.__foreign_dao[fdao].__foreign_tables.values()
],
)
else:
if attr in self.__date_attributes or String.to_snake_case(attr) in self.__date_attributes:
attr = self._attr_from_date_to_char(f"{self._table_name}.{attr}")
builder.with_value_condition(
attr,
operator,
self._get_value_sql(value),
[
x[0]
for fdao in self.__foreign_dao
for x in self.__foreign_dao[fdao].__foreign_tables.values()
],
)
def _graphql_to_sql_conditions(
self, graphql_structure: dict, external_table_deps: list[str]
) -> list[tuple[str, str, Any]]:
"""
Converts a GraphQL-like structure to SQL conditions.
:param graphql_structure: The GraphQL-like filter structure.
:param external_table_deps: List to track external table dependencies.
:return: A list of tuples (attribute, operator, value).
"""
operators = {
"equal": "=",
"notEqual": "!=",
"greater": ">",
"greaterOrEqual": ">=",
"less": "<",
"lessOrEqual": "<=",
"isNull": "IS NULL",
"isNotNull": "IS NOT NULL",
"contains": "LIKE", # Special handling in _graphql_to_sql_conditions
"notContains": "NOT LIKE", # Special handling in _graphql_to_sql_conditions
"startsWith": "LIKE", # Special handling in _graphql_to_sql_conditions
"endsWith": "LIKE", # Special handling in _graphql_to_sql_conditions
"in": "IN",
"notIn": "NOT IN",
}
conditions = []
def parse_node(node, parent_key=None, parent_dao=None):
if not isinstance(node, dict):
return
if isinstance(node, list):
conditions.append((parent_key, "IN", node))
return
for key, value in node.items():
if isinstance(key, property):
key = key.fget.__name__
external_fields_table_name_by_parent = self._get_external_field_key(parent_key)
external_fields_table_name = self._get_external_field_key(key)
external_field = (
external_fields_table_name
if external_fields_table_name_by_parent is None
else external_fields_table_name_by_parent
)
if key == "fuzzy":
self._handle_fuzzy_filter_conditions(conditions, external_table_deps, value)
elif parent_dao is not None and key in parent_dao.__db_names:
parse_node(value, f"{parent_dao.table_name}.{key}")
continue
elif external_field is not None:
external_table_deps.append(external_field)
parse_node(value, f"{external_field}.{key}")
elif parent_key in self.__foreign_table_keys:
if key in operators:
parse_node({key: value}, self.__foreign_table_keys[parent_key])
continue
if parent_key in self.__foreign_dao:
foreign_dao = self.__foreign_dao[parent_key]
if key in foreign_dao.__foreign_tables:
parse_node(
value,
f"{self.__foreign_tables[parent_key][0]}.{foreign_dao.__foreign_table_keys[key]}",
foreign_dao.__foreign_dao[key],
)
continue
if parent_key in self.__foreign_tables:
parse_node(value, f"{self.__foreign_tables[parent_key][0]}.{key}")
continue
parse_node({parent_key: value})
elif key in operators:
operator = operators[key]
if key == "contains" or key == "notContains":
value = f"%{value}%"
elif key == "in" or key == "notIn":
value = value
elif key == "startsWith":
value = f"{value}%"
elif key == "endsWith":
value = f"%{value}"
elif key == "isNull" or key == "isNotNull":
is_null_value = value.get("equal", None) if isinstance(value, dict) else value
if is_null_value is None:
operator = operators[key]
elif (key == "isNull" and is_null_value) or (key == "isNotNull" and not is_null_value):
operator = "IS NULL"
else:
operator = "IS NOT NULL"
conditions.append((parent_key, operator, None))
elif (key == "equal" or key == "notEqual") and value is None:
operator = operators["isNull"]
conditions.append((parent_key, operator, value))
elif isinstance(value, dict):
if key in self.__foreign_table_keys:
parse_node(value, key)
elif key in self.__db_names and parent_key is not None:
parse_node({f"{parent_key}": value})
elif key in self.__db_names:
parse_node(value, self.__db_names[key])
else:
parse_node(value, key)
elif value is None:
conditions.append((self.__db_names[key], "IS NULL", value))
else:
conditions.append((self.__db_names[key], "=", value))
parse_node(graphql_structure)
return conditions
def _handle_fuzzy_filter_conditions(self, conditions, external_field_table_deps, sub_values):
# Extract fuzzy filter parameters
fuzzy_fields = get_value(sub_values, "fields", list[str])
fuzzy_term = get_value(sub_values, "term", str)
fuzzy_threshold = get_value(sub_values, "threshold", int, 5)
if not fuzzy_fields or not fuzzy_term:
raise ValueError("Fuzzy filter must include 'fields' and 'term'.")
fuzzy_fields_db_names = []
# Map fields to their database names
for fuzzy_field in fuzzy_fields:
external_fields_table_name = self._get_external_field_key(fuzzy_field)
if external_fields_table_name is not None:
external_fields_table = self._external_fields[external_fields_table_name]
fuzzy_fields_db_names.append(f"{external_fields_table.table_name}.{fuzzy_field}")
external_field_table_deps.append(external_fields_table.table_name)
elif fuzzy_field in self.__db_names:
fuzzy_fields_db_names.append(f"{self._table_name}.{self.__db_names[fuzzy_field]}")
elif fuzzy_field in self.__foreign_tables:
fuzzy_fields_db_names.append(f"{self._table_name}.{self.__foreign_table_keys[fuzzy_field]}")
else:
fuzzy_fields_db_names.append(self.__db_names[String.to_snake_case(fuzzy_field)][0])
# Build fuzzy conditions for each field
fuzzy_conditions = self._build_fuzzy_conditions(fuzzy_fields_db_names, fuzzy_term, fuzzy_threshold)
# Combine conditions with OR and append to the main conditions
conditions.append((f"({' OR '.join(fuzzy_conditions)})", "fuzzy", None))
@staticmethod
def _build_fuzzy_conditions(fields: list[str], term: str, threshold: int = 10) -> list[str]:
conditions = []
for field in fields:
conditions.append(f"levenshtein({field}::TEXT, '{term}') <= {threshold}") # Adjust the threshold as needed
return conditions
def _get_external_field_key(self, field_name: str) -> Optional[str]:
"""
Returns the key to get the external field if found, otherwise None.
:param str field_name: The name of the field to search for.
:return: The key if found, otherwise None.
:rtype: Optional[str]
"""
if field_name is None:
return None
for key, builder in self._external_fields.items():
if field_name in builder.fields and field_name not in self.__db_names:
return key
return None
def _get_recursive_reference_join(self, attr: str) -> Optional[tuple[str, str]]:
parts = attr.split(".")
table_name = ".".join(parts[:-1])
if table_name == self._table_name or table_name == "":
return None
all_foreign_tables = {
x[0]: x[1]
for x in [
*[x for x in self.__foreign_tables.values() if x[0] != self._table_name],
*[x for fdao in self.__foreign_dao for x in self.__foreign_dao[fdao].__foreign_tables.values()],
]
}
if not table_name in all_foreign_tables:
return None
return table_name, all_foreign_tables[table_name]
def _build_sorts(
self,
builder: SQLSelectBuilder,
sorts: AttributeSorts,
external_table_deps: list[str],
):
"""
Resolves complex sorting structures into SQL-compatible sorting conditions.
Tracks external table dependencies.
:param builder: The SQLBuilder instance to add sorting to.
:param sorts: Sorting attributes and directions in a complex structure.
:param external_table_deps: List to track external table dependencies.
"""
def parse_sort_node(node, parent_key=None):
if isinstance(node, dict):
for key, value in node.items():
if isinstance(value, dict):
# Recursively parse nested structures
parse_sort_node(value, key)
elif isinstance(value, str) and value.lower() in ["asc", "desc"]:
external_table = self._get_external_field_key(key)
if external_table:
external_table_deps.append(external_table)
key = f"{external_table}.{key}"
if parent_key in self.__foreign_tables:
key = f"{self.__foreign_tables[parent_key][0]}.{key}"
builder.with_order_by(key, value.upper())
else:
raise ValueError(f"Invalid sort direction: {value}")
elif isinstance(node, list):
for item in node:
parse_sort_node(item)
else:
raise ValueError(f"Invalid sort structure: {node}")
parse_sort_node(sorts)
def _get_value_sql(self, value: Any) -> str:
if isinstance(value, str):
if value.lower() == "null":
return "NULL"
return f"'{value}'"
if isinstance(value, NoneType):
return "NULL"
if value is None:
return "NULL"
if isinstance(value, Enum):
return f"'{value.value}'"
if isinstance(value, bool):
return "true" if value else "false"
if isinstance(value, list):
if len(value) == 0:
return "()"
return f"({', '.join([self._get_value_sql(x) for x in value])})"
if isinstance(value, datetime.datetime):
if value.tzinfo is None:
value = value.replace(tzinfo=datetime.timezone.utc)
return f"'{value.strftime(DATETIME_FORMAT)}'"
return str(value)
@staticmethod
def _get_value_from_sql(cast_type: type, value: Any) -> Optional[T]:
"""
Get the value from the query result and cast it to the correct type
:param type cast_type:
:param Any value:
:return Optional[T]: Casted value, when value is str "NULL" None is returned
"""
if isinstance(value, str) and "NULL" in value:
return None
if isinstance(value, NoneType):
return None
if isinstance(value, cast_type):
return value
return cast_type(value)
def _get_primary_key_value_sql(self, obj: T_DBM) -> str:
value = getattr(obj, self.__primary_key)
if isinstance(value, str):
return f"'{value}'"
return value
@staticmethod
def _attr_from_date_to_char(attr: str) -> str:
return f"TO_CHAR({attr}, 'YYYY-MM-DD HH24:MI:SS.US TZ')"
@staticmethod
async def _get_editor_id(obj: T_DBM):
editor_id = obj.editor_id
if editor_id is None:
from cpl.core.ctx.user_context import get_user
user = get_user()
if user is not None:
editor_id = user.id
return editor_id if editor_id is not None else "NULL"

Some files were not shown because too many files have changed in this diff Show More