fix: MR fix

Dockerfile

@@ -20,7 +20,7 @@ RUN apk update --no-cache \
     && apk add --no-cache git
 
 COPY --from=builder --chown=app:app /app/.venv /app/.venv
-COPY --from=builder --chown=app:app /app/app /app/app
+COPY --from=builder --chown=app:app /app/src /app/src
 
 ENV PYTHONPATH="/app"
 
@@ -28,6 +28,6 @@ EXPOSE 8081
 
 WORKDIR /app
 
-ENTRYPOINT ["app/.venv/bin/python"]
+ENTRYPOINT ["/app/.venv/bin/python"]
 
-CMD ["/app/app/main.py"]
+CMD ["/app/src/karl/__init__.py"]

app/api/models.py

@@ -1,16 +0,0 @@
-from dataclasses import dataclass
-from typing import List
-
-
-@dataclass
-class Request:
-  _id: str
-  commit: str
-  message: str
-  started: str
-  files: List[str]
-
-
-@dataclass
-class Response:
-  status: int

app/api/v1.py

@@ -1,38 +0,0 @@
-from automapper import mapper, exceptions
-from fastapi import APIRouter, Depends
-from fastapi_utils.cbv import cbv
-from starlette.responses import JSONResponse, Response
-
-from app.api.models import Request
-from app.core.injects import AutowireSupport
-from app.core.woodpecker import Woodpecker
-from app.model.webhook import WoodpeckerEvent
-
-router = APIRouter()
-
-
-@router.get("/", summary="Main API")
-async def root():
-  return {"message": "Witaj w API v1"}
-
-
-@cbv(router)
-class APIv1:
-  woodpecker: Woodpecker = Depends(AutowireSupport.woodpecker)
-  logger = __import__('logging').getLogger(__name__)
-
-  def __init__(self):
-    try:  # TODO: rejestracja w innym miejscu: klasa jest przeładowywana co żądanie
-      mapper.add(Request, WoodpeckerEvent)
-    except exceptions.DuplicatedRegistrationError:
-      pass
-
-  @router.get("/health", summary="Health check")
-  async def health(self) -> JSONResponse:
-    # TODO: JSON serialize
-    return JSONResponse({"status": "unknown"})
-
-  @router.post("/ci", summary="CI Webhook")
-  async def ci(self, request: Request):
-    self.woodpecker.on_ci_event(mapper.map(request))
-    return Response(status_code=201)

app/config/__init__.py

@@ -1,7 +0,0 @@
-from .settings import AppConfig
-from .settings import GitConfig
-from .settings import KeePassConfig
-from .settings import Settings
-from .settings import get_settings
-
-__all__ = [AppConfig, GitConfig, KeePassConfig, Settings, get_settings]

app/config/settings.py

@@ -1,54 +0,0 @@
-from functools import lru_cache
-from pathlib import Path
-
-import yaml
-from pydantic import BaseModel
-from pydantic_settings import BaseSettings, SettingsConfigDict
-
-
-class LoggingConfig(BaseModel):
-  level: str = "INFO"
-  path: Path = Path("logs/karl.log")
-
-
-class AppConfig(BaseModel):
-  host: str = "127.0.0.1"
-  port: int = 8081
-  reload: bool = False
-
-
-class GitConfig(BaseModel):
-  path: Path = Path("/opt/repo/sample")
-  url: str = "ssh://git@hattori.ztsh.eu:29418/paas/heimdall.git"
-  branch: str = "master"
-  remote: str = "origin"
-
-
-class KeePassConfig(BaseModel):
-  file: str = "database.kdbx"
-  secret: Path | str = "/run/secrets/kp_secret"
-
-
-class Settings(BaseSettings):
-  model_config = SettingsConfigDict(env_prefix="KARL_", env_nested_delimiter="__")
-  logging: LoggingConfig = LoggingConfig()
-  app: AppConfig = AppConfig()
-  git: GitConfig = GitConfig()
-  kp: KeePassConfig = KeePassConfig()
-
-  @classmethod
-  def from_yaml(cls, path: Path | str = "config/config.yaml") -> "Settings":
-    p = Path(path)
-    data = {}
-    if p.exists():
-      with p.open("r", encoding="utf-8") as fh:
-        data = yaml.safe_load(fh) or {}
-    else:
-      import sys
-      sys.stderr.write(f"Warning: Config file {p} not found.\n")
-    return cls(**data)
-
-
-@lru_cache
-def get_settings() -> Settings:
-  return Settings.from_yaml()

app/core/injects.py

@@ -1,10 +0,0 @@
-from injectable import inject
-
-from app.core.woodpecker import Woodpecker
-
-
-class AutowireSupport:
-
-  @staticmethod
-  def woodpecker():
-    return inject(Woodpecker)

app/core/router.py

@@ -1,20 +0,0 @@
-from fastapi import APIRouter, Request
-
-from fastapi.responses import HTMLResponse
-from jinja2 import Environment, FileSystemLoader, select_autoescape
-
-router = APIRouter()
-
-# Inicjalizacja Jinja2
-templates_env = Environment(
-  loader=FileSystemLoader("app/templates"),
-  autoescape=select_autoescape(["html", "xml"]),
-)
-
-
-# Przykładowy endpoint HTML
-@router.get("/", response_class=HTMLResponse)
-async def index(request: Request) -> HTMLResponse:
-  template = templates_env.get_template("index.html")
-  html = template.render(title="Strona główna", request=request)
-  return HTMLResponse(content=html)

app/core/woodpecker.py

@@ -1,107 +0,0 @@
-import logging
-from collections import deque
-from multiprocessing import Process, Lock
-from pathlib import Path
-from typing import Annotated
-
-from injectable import injectable, Autowired, autowired
-
-from app.config import get_settings
-from app.model.webhook import WoodpeckerEvent
-from app.services import GitService, DockerService
-from app.services.mo import Mo
-
-logger = logging.getLogger(__name__)
-
-
-class WoodpeckerRunner(Process):
-  def __init__(self, git: GitService, docker: DockerService, mo: Mo,
-               success_callback=None, error_callback=None):
-    super().__init__(daemon=True)
-    self._git = git
-    self._docker = docker
-    self._mo = mo
-    self._success_callback = success_callback
-    self._error_callback = error_callback
-    self._event: WoodpeckerEvent | None = None
-    self._root = get_settings().git.path
-
-  def process_event(self, event: WoodpeckerEvent):
-    self._event = event
-    self.start()
-
-  def run(self):
-    try:
-      service = self.get_service(self._event.files)
-      if service is None:
-        logger.info("No service found.")
-        return self._success_callback()
-      service_path = f"{self._root}/compose/{service}/docker-compose.yml"
-      self._git.checkout(self._event.commit)
-      for file in self._event.files:
-        if file.__contains__('.mo.'):
-          self._mo.process(Path(f"{self._root}{file}").absolute())
-      self._docker.reload(Path(service_path).absolute())
-
-      return self._success_callback()
-    except Exception as e:
-      return self._error_callback(e)
-
-  def get_service(self, files: list[str]) -> str | None:
-    supported_files = []
-    for f in files:
-      f_parts = f.split("/")
-      if f_parts[0] in ["compose", "files"]:
-        supported_files.append(f[1])
-    match len(set(supported_files)):
-      case 0:
-        return None
-      case 1:
-        return supported_files[0]
-      case _:
-        raise Exception("Multiple services are not supported.")
-
-
-@injectable(singleton=True)
-class Woodpecker:
-  @autowired
-  def __init__(self, mo: Annotated[Mo, Autowired]):
-    self._mo = mo
-    self._git = GitService()
-    self._docker = DockerService()
-    self._runner: WoodpeckerRunner | None = None
-    self._pending = deque()
-    self._lock = Lock()
-    logger.info("Woodpecker initialized.")
-
-  def on_ci_event(self, event: WoodpeckerEvent):
-    logger.info(f"Received event: {event}")
-    with self._lock:
-      if len(self._pending) > 0 or self._runner is not None:
-        self._pending.append(event)
-        return
-      self._start_runner(event)
-
-  def _start_runner(self, event: WoodpeckerEvent):
-    with self._lock:
-      self._runner = WoodpeckerRunner(self._git, self._docker, self._mo,
-                                      self._on_runner_completed, self._on_runner_error)
-      self._runner.process_event(event)
-
-  def _on_runner_completed(self):
-    logger.info("Runner completed.")
-    self._runner.join()
-    with self._lock:
-      self._runner = None
-      if len(self._pending) > 0:
-        event = self._pending.popleft()
-        self._start_runner(event)
-
-  def _on_runner_error(self, t: Exception):
-    logger.error(f"Runner error: {t}", exc_info=True)
-    self._runner.join()
-    with self._lock:
-      self._runner = None
-      if len(self._pending) > 0:
-        event = self._pending.popleft()
-        self._start_runner(event)

app/model/containers.py

@@ -1,40 +0,0 @@
-from dataclasses import dataclass, field
-from datetime import datetime
-
-from docker.models.containers import Container
-
-
-@dataclass
-class SimpleContainer:
-  name: str
-  image: str
-  status: str
-  health: str
-  created: datetime
-
-  @staticmethod
-  def from_container(container: Container):
-    created = datetime.strptime(container.attrs['Created'].split('.')[0], '%Y-%m-%dT%H:%M:%S')
-    return SimpleContainer(
-      name=container.name,
-      image=container.image.tags[0],
-      status=container.status,
-      health=container.health,
-      created=created
-    )
-
-
-@dataclass
-class Compose:
-  directory: str
-  containers: list[SimpleContainer] = field(default_factory=list)
-
-  @property
-  def last_modified(self):
-    return max(self.containers, key=lambda c: c.created).created
-
-
-@dataclass
-class Tree:
-  composes: dict[str, Compose] = field(default_factory=dict)
-  containers: list[SimpleContainer] = field(default_factory=list)

app/model/healthcheck.py

@@ -1,7 +0,0 @@
-from dataclasses import dataclass
-
-
-@dataclass
-class HealthCheck:
-  healthy: bool
-  message: str

app/model/passwords.py

@@ -1,52 +0,0 @@
-from dataclasses import dataclass, field
-from typing import Type
-
-
-# TODO: unnecessary?
-
-@dataclass
-class PathItem:
-  name: str
-  t: Type
-
-
-@dataclass
-class Path:
-  path: list[PathItem] = field(default_factory=list)
-
-  def append(self, name, t):
-    self.path.append(PathItem(name, t))
-
-  def __str__(self):
-    return "/".join([i.name for i in self.path])
-
-
-@dataclass
-class Group:
-  name: str
-  passwords: list["Password"] = field(default_factory=list)
-  parent: "Group|None" = None
-
-  @property
-  def path(self):
-    if self.parent is None:
-      new_path = Path()
-      new_path.append(self.name, type(self))
-      return new_path
-    return self.parent.path.append(self.name, type(self))
-
-
-@dataclass
-class Password:
-  name: str
-  group: Group
-
-  @property
-  def path(self):
-    return self.group.path.append(self.name, type(self))
-
-
-class UnencryptedPassword(Password):
-  def __init__(self, name: str, value: str, group: Group):
-    super().__init__(name, group)
-    self.value = value

app/model/webhook.py

@@ -1,11 +0,0 @@
-from dataclasses import dataclass
-from typing import List
-
-
-@dataclass
-class WoodpeckerEvent:
-  _id: str
-  commit: str
-  message: str
-  started: str
-  files: List[str]

app/services/__init__.py

@@ -1,5 +0,0 @@
-from .containers import DockerService
-from .passwords import Passwords
-from .vcs import GitService
-
-__all__ = ["GitService", "Passwords", "DockerService"]

app/services/containers.py

@@ -1,59 +0,0 @@
-import logging
-from pathlib import Path
-
-import docker
-from docker.models.containers import Container
-from injectable import injectable
-
-from app.model.containers import Tree, Compose, SimpleContainer
-
-logger = logging.getLogger(__name__)
-
-
-@injectable(singleton=True)
-class DockerService:
-  def __init__(self):
-    self._client = docker.from_env()
-    # logger.info(f"Docker client initialized. Plugins: {self._client.plugins()}")
-    self._tree = self._init_tree()
-
-  def _init_tree(self) -> Tree:
-    tree = Tree()
-    container: Container
-    for container in self._client.containers.list():
-      labels = container.labels
-      working_dir = labels.get("com.docker.compose.project.working_dir")
-      if working_dir:
-        if tree.composes.get(working_dir) is None:
-          tree.composes[working_dir] = Compose(working_dir)
-        tree.composes[working_dir].containers.append(SimpleContainer.from_container(container))
-      else:
-        tree.containers.append(SimpleContainer.from_container(container))
-    return tree
-
-  @property
-  def tree(self) -> Tree:
-    return self._tree
-
-  def reload(self, compose_path: Path):
-    # TODO: Won't work in docker container
-    cmd = ["sudo", "docker", "compose", "-f", str(compose_path), "up", "-d"]
-    import subprocess
-    try:
-      process = subprocess.run(
-        cmd,
-        capture_output=True,
-        text=True,
-        check=False
-      )
-      if process.returncode != 0:
-        logger.error(f"Docker compose failed with code {process.returncode}")
-        logger.error(f"stderr: {process.stderr}")
-        raise Exception(f"Docker compose failed: {process.stderr}")
-
-      logger.info(f"Docker compose executed successfully")
-      logger.debug(f"stdout: {process.stdout}")
-      return process.stdout, process.stderr, process.returncode
-    except Exception as e:
-      logger.error(f"Failed to execute docker compose command: {e}")
-      raise e

app/services/mo.py

@@ -1,30 +0,0 @@
-from pathlib import Path
-from string import Template
-from typing import Annotated
-
-from injectable import injectable, autowired, Autowired
-
-from app.services import Passwords
-
-
-class ValueTemplate(Template):
-  # Pozwala na kropki i ukośniki w nazwach placeholderów, np. ${user.name/first}
-  idpattern = r'[_a-zA-Z][_a-zA-Z0-9.\/]*'
-
-
-@injectable
-class Mo:
-  @autowired
-  def __init__(self, passwords: Annotated[Passwords, Autowired]):
-    self._passwords = passwords
-
-  def process(self, mo_file: Path):
-    raw = ''
-    with open(mo_file, "r") as mo:
-      raw = mo.read()
-    parsed = ValueTemplate(raw)
-    mappings = self._passwords.get_values(parsed.get_identifiers())
-    rendered = parsed.safe_substitute(mappings)
-    de_mo_ified = str(mo_file).replace(".mo", "")
-    with open(de_mo_ified, "w") as mo:
-      mo.write(rendered)

app/services/passwords.py

@@ -1,73 +0,0 @@
-import os.path
-from contextlib import contextmanager
-from typing import Any, Generator
-
-import keyring
-from injectable import injectable
-from pykeepass import PyKeePass, create_database
-
-class KeyRequest:
-  def __init__(self, prompt: str):
-    self.field_name = None
-    self.entry_name = None
-    self.path = None
-    self._parse_prompt(prompt)
-
-  def _parse_prompt(self, prompt: str):
-    prompt_parts = prompt.split("/")
-    key = None
-    match len(prompt_parts):
-      case 1:
-        self.field_name = 'password'
-        key = prompt_parts[0]
-      case 2:
-        self.field_name = prompt_parts[1]
-        key = prompt_parts[0]
-      case _:
-        key = None
-    if key is None:
-      return
-    key_parts = key.split(".")
-    self.path = key_parts[:] if len(key_parts) > 1 else None
-    self.entry_name = key_parts[-1]
-
-@injectable(singleton=True)
-class Passwords:
-  def __init__(self):
-    from app.config import get_settings
-    settings = get_settings()
-
-    with open(settings.kp.secret, "r") as fh:
-      keyring.set_password("karl", "kp", fh.read().splitlines()[0])
-    self._path = settings.kp.file
-
-  @contextmanager
-  def open(self, mode: str = "r") -> Generator[PyKeePass | Any, Any, None]:
-    kp = PyKeePass(self._path, password=keyring.get_password("karl", "kp")) \
-      if os.path.exists(self._path) else create_database(self._path, password=keyring.get_password("karl", "kp"))
-    yield kp
-    if mode == "rw":
-      kp.save()
-
-  def get_values(self, keys: list[str]) -> dict[str, str]:
-    output = {}
-    for k in keys:
-      request = KeyRequest(k)
-      with self.open() as kp:
-        kp_entry = kp.find_entries(path=request.path, first=True, title=request.entry_name)
-      output[k] = self._get_field_value(kp_entry, request.field_name)
-    return output
-
-  @staticmethod
-  def _get_field_value(kp_entry, field_name):
-    if kp_entry is None:
-      return None
-    match field_name:
-      case "username":
-        return kp_entry.username
-      case "password":
-        return kp_entry.password
-      case "url":
-        return kp_entry.url
-      case _:
-        return kp_entry.get_custom_property(field_name)

app/services/system.py

@@ -1 +0,0 @@
-

app/services/vcs.py

@@ -1,31 +0,0 @@
-from git import Repo, Remote
-from injectable import injectable
-
-from app.config import GitConfig, get_settings
-
-
-@injectable(singleton=True)
-class GitService:
-  def __init__(self):
-    self._settings = get_settings()
-    self._repo = self._check_preconditions(self._settings.git)
-    if self._repo.head.ref.name != self._settings.git.branch:
-      self._repo.git.checkout(self._settings.git.branch)
-    self._origin: Remote = self._repo.remotes.origin
-
-  @staticmethod
-  def _check_preconditions(config: GitConfig) -> Repo:
-    def clone():
-      return Repo.clone_from(config.url, config.path, branch=config.branch)
-
-    import os
-    if not config.path.exists():
-      return clone()
-    if not (config.path / ".git").exists():
-      os.rmdir(config.path)
-      return clone()
-    return Repo(config.path)
-
-  def checkout(self, sha: str):
-    self._origin.fetch()
-    self._repo.git.checkout(sha)

app/util/dicts.py

@@ -1,11 +0,0 @@
-from types import SimpleNamespace
-
-
-class NestedNamespace(SimpleNamespace):
-  def __init__(self, dictionary, **kwargs):
-    super().__init__(**kwargs)
-    for key, value in dictionary.items():
-      if isinstance(value, dict):
-        self.__setattr__(key, NestedNamespace(value))
-      else:
-        self.__setattr__(key, value)

app/util/logging.py

@@ -1,90 +0,0 @@
-from enum import Enum, auto
-from logging import Formatter, StreamHandler, Handler
-from logging.handlers import TimedRotatingFileHandler
-from pathlib import Path
-from typing import List
-
-
-class NamingCache:
-  def __init__(self):
-    self._cache = {}
-
-  def __getitem__(self, key):
-    if key not in self._cache:
-      self._cache[key] = self.shorten(key)
-    return self._cache[key]
-
-  @staticmethod
-  def shorten(logger_name: str) -> str:
-    target_length = 18
-    if len(logger_name) > target_length:
-      parts = logger_name.split('.')
-      part = 0
-      while len(logger_name) > target_length:
-        if part == len(parts) - 1:
-          logger_name = f'...{logger_name[-(target_length - 3):]}'
-          break
-        parts[part] = parts[part][0]
-        logger_name = '.'.join(parts)
-        part += 1
-
-    return logger_name.ljust(target_length)
-
-
-class ApplicationFormatter(Formatter):
-  def __init__(self, handler_prefix: str = ''):
-    super().__init__()
-    self._logger_names = NamingCache()
-    self._handler_prefix = handler_prefix
-
-  def format(self, record):
-    from datetime import datetime
-    timestamp = datetime.fromtimestamp(record.created).isoformat(sep=' ', timespec='milliseconds')
-    level = record.levelname.replace('WARNING', 'WARN').rjust(5)
-    thread_name = record.threadName.replace(' (', '#').replace(')', '').rjust(16)[-16:]  # TODO: NamingCache?
-    logger_name = self._logger_names[f"{self._handler_prefix}{record.name}"]
-    message = record.getMessage()
-    formatted = f"{timestamp} {level} [{thread_name}] {logger_name} : {message}"
-
-    if record.exc_info:
-      formatted += "\n" + self.formatException(record.exc_info)
-
-    return formatted
-
-
-class HandlerFactory:
-  class Target(Enum):
-    CONSOLE = auto()
-    FILE = auto()
-    ALL = auto()
-
-  @staticmethod
-  def create(target: Target, handler_prefix: str = '', file_path: Path = None) -> List[Handler]:
-    def console_handler(prefix: str = ''):
-      handler = StreamHandler()
-      handler.setFormatter(ApplicationFormatter(prefix))
-      handler.setLevel('INFO')
-      return handler
-
-    def file_handler(prefix: str = ''):
-      if not file_path:
-        raise ValueError("File path must be set.")
-      file_path.parent.mkdir(parents=True, exist_ok=True)
-      file_path.touch(exist_ok=True)
-      handler = TimedRotatingFileHandler(file_path, when='midnight', backupCount=30)
-      handler.setFormatter(ApplicationFormatter(prefix))
-      handler.setLevel('TRACE')
-      return handler
-
-    handlers = []
-    match target:
-      case HandlerFactory.Target.CONSOLE:
-        handlers.append(console_handler(handler_prefix))
-      case HandlerFactory.Target.FILE:
-        handlers.append(file_handler(handler_prefix))
-      case HandlerFactory.Target.ALL:
-        handlers.append(file_handler(handler_prefix))
-        handlers.append(console_handler(handler_prefix))
-      case _:
-        raise ValueError(f"Unknown target: {target}")
-    return handlers

app/web/middlewares.py

@@ -1,24 +0,0 @@
-import logging
-
-from starlette.middleware.base import BaseHTTPMiddleware
-from starlette.requests import Request
-
-logger = logging.getLogger(__name__)
-
-
-class LoggingMiddleware(BaseHTTPMiddleware):
-  async def dispatch(self, request: Request, call_next):
-    client = f"{request.client.host}:{request.client.port}"
-    match request.method:
-      case "POST" | "PUT" | "DELETE" if request.headers.get("Content-Type") == "application/json":
-        body = await request.body()
-        logger.trace(f"Request from {client}: {body.decode()}")
-      case "GET":
-        logger.trace(f"Request from {client}")
-      case _:
-        logger.trace(f"Request from {client} (content-type:{request.headers.get("Content-Type")})")
-
-    response = await call_next(request)
-    logger.trace(f"Respone: {response.status_code} {type(response)}")
-
-    return response