Reload

.dockerignore

@@ -0,0 +1,62 @@
+# JB Toolkit
+.idea
+
+# Git
+.git
+.gitignore
+.gitattributes
+
+# CI
+.woodpecker/
+
+# Virtual env
+.venv/
+
+# Docker
+docker-compose.yml
+Dockerfile
+.docker
+.dockerignore
+
+# Byte-compiled / optimized / DLL files
+**/__pycache__/
+**/*.py[cod]
+
+# C extensions
+*.so
+
+# Distribution / packaging
+.Python
+env/
+build/
+develop-eggs/
+dist/
+downloads/
+target/
+eggs/
+lib/
+lib64/
+parts/
+sdist/
+var/
+*.egg-info/
+.installed.cfg
+*.egg
+
+# Logs
+**/*.log
+
+# Documentation
+docs/_build/
+
+# Python mode for VIM
+.ropeproject
+**/.ropeproject
+
+# Vim swap files
+**/*.swp
+
+# Project specific
+.compose_repository/
+config/
+tests/

.editorconfig

@@ -8,3 +8,6 @@ trim_trailing_whitespace = true
 
 [*.py]
 indent_size = 2
+
+[*.yaml]
+indent_size = 2

.gitignore

@@ -1,6 +1,19 @@
+# Python-generated files
+__pycache__/
+*.py[oc]
+build/
+dist/
+wheels/
+*.egg-info
+
+# Virtual environments
+.venv
 .idea
 *.iml
 uv.lock
+**/*.kdbx*
+.compose_repository
 
 __pycache__/
 **/dist/
+**/*.log

.python-version

@@ -0,0 +1 @@
+3.12

.woodpecker/dev.yaml

@@ -0,0 +1,14 @@
+steps:
+  - name: build
+    image: woodpeckerci/plugin-docker-buildx:5.2.2
+    settings:
+      platforms: linux/amd64
+      repo: hattori.ztsh.eu/iac/karl
+      registry: hattori.ztsh.eu
+      tags: dev-${CI_PIPELINE_NUMBER}
+      username: stawros
+      password:
+        from_secret: hattori-packages
+    when:
+      - event: pull_request
+        evaluate: 'CI_COMMIT_PULL_REQUEST_LABELS contains "ci-ready"'

.woodpecker/latest.yaml

@@ -0,0 +1,14 @@
+steps:
+  - name: build
+    image: woodpeckerci/plugin-docker-buildx:5.2.2
+    settings:
+      platforms: linux/amd64
+      repo: hattori.ztsh.eu/iac/karl
+      registry: hattori.ztsh.eu
+      tags: latest
+      username: stawros
+      password:
+        from_secret: hattori-packages
+    when:
+      - event: [ tag, push, manual ]
+        branch: master

Dockerfile

@@ -0,0 +1,33 @@
+FROM ghcr.io/astral-sh/uv:0.9-python3.12-alpine AS builder
+
+WORKDIR /app
+
+RUN apk update \
+    && apk add gcc python3-dev musl-dev linux-headers
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    --mount=type=bind,source=pyproject.toml,target=pyproject.toml \
+    uv sync --no-install-workspace
+
+ADD . /app
+
+RUN --mount=type=cache,target=/root/.cache/uv \
+    uv sync --locked
+
+FROM python:3.12-alpine3.22
+
+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
+
+ENV PYTHONPATH="/app"
+
+EXPOSE 8081
+
+WORKDIR /app
+
+ENTRYPOINT ["app/.venv/bin/python"]
+
+CMD ["/app/app/main.py"]

app/__main__.py

@@ -1,4 +0,0 @@
-if __name__ == '__main__':
-  from main import run
-
-  run()

app/api/v1.py

@@ -1,8 +0,0 @@
-from fastapi import APIRouter
-
-router = APIRouter()
-
-
-@router.get("/", summary="Main API")
-async def root():
-  return {"message": "Witaj w API v12"}

config/config.yaml

@@ -0,0 +1,13 @@
+logging:
+    level: "TRACE"
+    path: "logs/karl.log"
+app:
+    host: "127.0.0.1"
+    port: 8081
+    reload: false
+git:
+    path: "F:/IdeaProjects/paas/karl/.compose_repository"
+    branch: "main"
+kp:
+    file: "config/kp.kdbx"
+    secret: "config/secret.txt"

config/secret.txt

@@ -0,0 +1 @@
+supersecret

docker-compose.yaml

@@ -0,0 +1,16 @@
+services:
+  karl:
+    image: hattori.ztsh.eu/iac/karl:dev-17
+    container_name: karl
+    restart: always
+    networks:
+      - web
+    volumes:
+      - /etc/localtime:/etc/localtime:ro
+      - /var/run/docker.sock:/var/run/docker.sock
+      - ./data/config:/app/config
+      - ./data/logs:/app/logs
+
+networks:
+  web:
+    external: true

pyproject.toml

@@ -5,28 +5,38 @@ description = "Because name 'Jenkins' was already taken. Greatest composer ever.
 readme = "README.md"
 requires-python = ">=3.12"
 authors = [{ name = "Piotr Dec" }]
+
 dependencies = [
-    "fastapi>=0.115.0",
+    "fastapi>=0.119.0",
     "uvicorn[standard]>=0.30.0",
     "jinja2>=3.1.4",
+    "pydantic-settings>=2.4.0",
+    "pyyaml>=6.0.2",
+    "gitpython>=3.1.45",
+    "pykeepass>=4.1.1.post1",
+    "docker>=7.1.0",
+    "injectable==4.0.1",
+    "py-automapper>=2.2.0",
+    "fastapi-utils>=0.8.0",
+    "keyring>=25.6.0",
 ]
 
-[project.optional-dependencies]
+[dependency-groups]
 dev = [
     "httpx>=0.27.0",
-    "pytest>=8.3.0",
-    "pytest-asyncio>=0.23.0",
+    "pytest==9.0.1",
+    "pytest-asyncio>=1.3.0",
     "ruff>=0.6.0",
     "mypy>=1.11.0",
     "types-Jinja2>=2.11.9",
 ]
 
 [project.scripts]
-app = "app.main:run"
+karl = "karl:main"
 
-[tool.uv]
-# uv automatycznie wykrywa dependencies z [project]
-# Możesz dodać tu własne ustawienia cache/mirrors, jeśli potrzebne.
+[build-system]
+requires = ["uv_build>=0.8.23,<0.9.0"]
+build-backend = "uv_build"
 
 [tool.ruff]
 line-length = 120

run.sh

@@ -1 +0,0 @@
-uv run uvicorn app.main:app --reload

src/karl/__init__.py

@@ -0,0 +1,2 @@
+def main() -> None:
+    print("Hello from karl!")

src/karl/api/models.py

@@ -0,0 +1,16 @@
+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

src/karl/api/v1.py

@@ -0,0 +1,38 @@
+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)

src/karl/config/__init__.py

@@ -0,0 +1,7 @@
+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]

src/karl/config/settings.py

@@ -0,0 +1,54 @@
+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()

src/karl/core/injects.py

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

src/karl/core/router.py

@@ -1,8 +1,9 @@
-from fastapi import FastAPI, Request
+from fastapi import APIRouter, Request
+
 from fastapi.responses import HTMLResponse
 from jinja2 import Environment, FileSystemLoader, select_autoescape
 
-from app.api.v1 import router as api_v1_router
+router = APIRouter()
 
 # Inicjalizacja Jinja2
 templates_env = Environment(
@@ -10,20 +11,10 @@ templates_env = Environment(
   autoescape=select_autoescape(["html", "xml"]),
 )
 
-app = FastAPI(title="Karl", version="0.1.0")
-
-# Rejestracja routera API pod /api/v1
-app.include_router(api_v1_router, prefix="/api/v1", tags=["v1"])
-
 
 # Przykładowy endpoint HTML
-@app.get("/", response_class=HTMLResponse)
+@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)
-
-
-def run() -> None:
-  import uvicorn
-  uvicorn.run("app.main:app", host="127.0.0.1", port=8000, reload=True)

src/karl/core/woodpecker.py

@@ -0,0 +1,107 @@
+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)

src/karl/main.py

@@ -0,0 +1,77 @@
+import logging
+
+from fastapi import FastAPI
+from injectable import load_injection_container
+
+from app.config import get_settings
+from app.util.logging import HandlerFactory
+
+
+class KarlApplication:
+  from starlette.types import Receive, Scope, Send
+  def __init__(self) -> None:
+    self._set_logging()
+    load_injection_container()
+    _app = FastAPI(title="Karl", version="0.1.0")
+    self._set_middlewares(_app)
+    self._set_routes(_app)
+    self._set_events(_app)
+
+    self._app = _app
+
+  async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+    await self._app.__call__(scope, receive, send)
+
+  def _set_logging(self):
+    settings = get_settings()
+    logging.addLevelName(5, "TRACE")
+    logging.Logger.trace = lambda s, msg, *args, **kwargs: s.log(5, msg, *args, **kwargs)
+    logging.basicConfig(level=settings.logging.level,
+                        handlers=HandlerFactory.create(HandlerFactory.Target.ALL, handler_prefix='karl.',
+                                                       file_path=settings.logging.path))
+
+    loggers = (
+      "uvicorn",
+      "uvicorn.access",
+      "uvicorn.error",
+      "fastapi",
+      "asyncio",
+      "starlette",
+    )
+    external_handlers = HandlerFactory.create(HandlerFactory.Target.ALL, file_path=settings.logging.path)
+    for logger_name in loggers:
+      logging_logger = logging.getLogger(logger_name)
+      logging_logger.handlers = external_handlers
+      logging_logger.propagate = False
+
+  def _set_middlewares(self, app: FastAPI):
+    from app.web.middlewares import LoggingMiddleware
+    app.add_middleware(LoggingMiddleware)
+
+  def _set_routes(self, app: FastAPI):
+    from app.core.router import router as core_router
+    app.include_router(core_router)
+    from app.api.v1 import router as api_v1_router
+    app.include_router(api_v1_router, prefix="/api/v1", tags=["v1"])
+    pass
+
+  def _set_events(self, app: FastAPI):
+    pass
+
+
+def run():
+  return KarlApplication()
+
+
+if __name__ == "__main__":
+  import uvicorn
+
+  settings = get_settings()
+  uvicorn.run(
+    "app.main:run",
+    factory=True,
+    host=settings.app.host,
+    port=settings.app.port,
+    reload=settings.app.reload,
+    log_config=None,
+  )

src/karl/model/containers.py

@@ -0,0 +1,40 @@
+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)

src/karl/model/healthcheck.py

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

src/karl/model/passwords.py

@@ -0,0 +1,52 @@
+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

src/karl/model/webhook.py

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

src/karl/services/__init__.py

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

src/karl/services/containers.py

@@ -0,0 +1,59 @@
+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

src/karl/services/mo.py

@@ -0,0 +1,30 @@
+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)

src/karl/services/passwords.py

@@ -0,0 +1,73 @@
+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)

src/karl/services/system.py

@@ -0,0 +1 @@
+

src/karl/services/vcs.py

@@ -0,0 +1,31 @@
+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)

src/karl/templates/index.html

@@ -3,7 +3,7 @@
 <head>
     <meta charset="utf-8"/>
     <title>{{ title }}</title>
-    <meta name="viewport" content="width=device-width, initial-scale=1"/>
+    <meta content="width=device-width, initial-scale=1" name="viewport"/>
     <style>
         body {
             font-family: system-ui, -apple-system, Segoe UI, Roboto, Ubuntu, Cantarell, Noto Sans, Helvetica, Arial, "Apple Color Emoji", "Segoe UI Emoji";

src/karl/util/dicts.py

@@ -0,0 +1,11 @@
+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)

src/karl/util/logging.py

@@ -0,0 +1,90 @@
+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

src/karl/web/middlewares.py

@@ -0,0 +1,24 @@
+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

tests/files/test1/test.mo.yaml

@@ -0,0 +1,5 @@
+value: ${sample}
+nested: ${some.nested.value}
+custom: ${custom/field}
+uname: ${sample/username}
+invalid: ${double/slash/example}

tests/test_mo.py

@@ -0,0 +1,50 @@
+from pathlib import Path
+
+import pytest
+import yaml
+
+from app.services import Passwords
+from app.services.mo import Mo
+
+
+@pytest.fixture(scope='class')
+def target_path():
+  p = Path('tests/files/test1/test.yaml')
+  # posprzątaj przed testem, gdyby plik istniał z poprzednich uruchomień
+  if p.exists():
+    p.unlink()
+  yield p
+  # sprzątanie po teście
+  if p.exists():
+    p.unlink()
+
+
+@pytest.fixture(scope='class')
+def test1_content(target_path: Path):
+  mo = Mo(Passwords())
+  mo.process(Path('tests/files/test1/test.mo.yaml').absolute())
+
+  assert target_path.exists()
+
+  content = target_path.read_text()
+  assert '${' not in content
+
+  yield yaml.load(content, Loader=yaml.FullLoader)
+
+
+class TestParsing:
+
+  def test_simple(self, test1_content: dict):
+    assert test1_content['value'] == 'some_pass'
+
+  def test_nested(self, test1_content: dict):
+    assert test1_content['nested'] == 'nested_pass'
+
+  def test_custom_field(self, test1_content: dict):
+    assert test1_content['custom'] == 'custom_content'
+
+  def test_username_field(self, test1_content: dict):
+    assert test1_content['uname'] == 'sample_username'
+
+  def test_invalid_key(self, test1_content: dict):
+    assert test1_content.get('invalid') == 'None'