alex/gitea-mirror
1
0
Fork 0
Code Issues Pull Requests Actions Projects Releases Wiki Activity

feat[logging]: flexible logging support

Browse Source
This commit is contained in:
Aleksey Lobanov
2026-06-28 00:48:50 +03:00
parent 206a8730a8
commit 17aaa41101
10 changed files with 403 additions and 12 deletions
Download Patch File Download Diff File Expand all files Collapse all files
config.example.toml
+23
View File
@@ -10,3 +10,26 @@ ssh_key = "/home/user/id_rsa"
# Use this fetch config instead of ssh_key to clone over HTTP(S):
# user = "gitea-user"
# token = "XXXXX"
[logging]
level = "INFO"
# Human-friendly logs in stdout.
[[logging.outputs]]
type = "console"
format = "plain"
stream = "stdout"
# Docker-friendly structured logs in stderr.
[[logging.outputs]]
type = "console"
format = "json"
stream = "stderr"
# Rotating file logs with selectable format.
[[logging.outputs]]
type = "file"
path = "/var/log/gitea-mirror/gitea-mirror.log"
format = "json"
max_size = 10485760
backup_count = 5
gitea-mirror.py
+42 -5
View File
@@ -1,3 +1,4 @@
import logging
import os.path
import sys
from os import makedirs
@@ -5,11 +6,20 @@ from os import makedirs
from src.config import Config, read_toml_config
from src.git import git_clone, git_pull, http_git_env, ssh_git_env
from src.gitea_api import GiteaApi
from src.log_fields import (
OPERATION,
PATH,
REPOSITORY,
REPOSITORY_COUNT,
REPOSITORY_URL,
)
from src.logging_config import setup_logging
from src.models import GiteaRepository
from src.repository_name import get_repository_name, is_valid_repository_names
BASE_PATH = "out"
FORMAT = "{owner}/{name}"
logger = logging.getLogger(__name__)
def get_fetch_params(config: Config, repo: GiteaRepository):
@@ -25,36 +35,63 @@ def process_repo(config: Config, repo: GiteaRepository):
path = get_repository_name(name_format=config.repository_format, r=repo)
out_path = os.path.join(config.out_dir, path)
repository_url, git_env = get_fetch_params(config=config, repo=repo)
log_context = {
REPOSITORY: path,
PATH: out_path,
REPOSITORY_URL: repository_url,
}
makedirs(out_path, exist_ok=True)
if os.path.exists(os.path.join(out_path, ".git")):
logger.info(
"repository pull started",
extra=log_context | {OPERATION: "pull"},
)
git_pull(out_path, env=git_env)
return
print(f"New repository: {path}")
logger.info(
"repository clone started",
extra=log_context | {OPERATION: "clone"},
)
git_clone(repository_url=repository_url, repository=out_path, env=git_env)
def main():
if len(sys.argv) < 2:
print("Usage: python gitea-mirror.py CONFIG_PATH")
logging.basicConfig(
level=logging.ERROR,
format="%(levelname)s %(message)s",
)
logger.error("Usage: python gitea-mirror.py CONFIG_PATH")
sys.exit(1)
try:
config = read_toml_config(sys.argv[1])
except RuntimeError as err:
print(f"Invalid config: {err}")
logging.basicConfig(
level=logging.ERROR,
format="%(levelname)s %(message)s",
)
logger.error("invalid config: %s", err)
sys.exit(1)
setup_logging(config.logging)
api = GiteaApi(
endpoint=config.endpoint,
api_token=config.api_token,
)
repos = api.get_repositories()
print(f"total {len(repos)} repositories")
logger.info(
"repositories ready",
extra={OPERATION: "mirror", REPOSITORY_COUNT: len(repos)},
)
if not is_valid_repository_names(
name_format=config.repository_format,
repos=repos,
):
print("Format string is not valid, duplicates are not allowed")
logger.error(
"repository name format creates duplicates",
extra={OPERATION: "validate_repository_names"},
)
sys.exit(1)
for repo in repos:
src/config.py
+3 -1
View File
@@ -17,7 +17,7 @@ ssh_key = "/home/user/.ssh/id_rsa"
import os
import tomllib
from .models import Config, FetchConfig
from .models import Config, FetchConfig, LoggingConfig
MAIN_SECTION = "main"
@@ -35,6 +35,7 @@ def read_toml_config(path: str) -> Config:
repository_format = main["format"]
out_dir = main["out_dir"]
fetch = FetchConfig(**data["fetch"])
logging_config = LoggingConfig(**data.get("logging", {}))
except (KeyError, ValueError, tomllib.TOMLDecodeError) as err:
raise RuntimeError(f"Invalid TOML config: {err}") from err
@@ -44,4 +45,5 @@ def read_toml_config(path: str) -> Config:
api_token=api_token,
out_dir=out_dir,
fetch=fetch,
logging=logging_config,
)
src/git.py
+24 -4
View File
@@ -1,9 +1,14 @@
import base64
import logging
import os
import shlex
import subprocess
from os import makedirs
from .log_fields import EXIT_CODE, OPERATION, PATH, REPOSITORY_URL
logger = logging.getLogger(__name__)
def ssh_git_env(ssh_key_path: str) -> dict[str, str]:
return {
@@ -37,8 +42,16 @@ def git_clone(
cwd=repository,
env=git_env,
)
except subprocess.CalledProcessError:
print(f"Unable to clone repository {repository} from {repository_url}")
except subprocess.CalledProcessError as err:
logger.error(
"git clone failed",
extra={
OPERATION: "clone",
PATH: repository,
REPOSITORY_URL: repository_url,
EXIT_CODE: err.returncode,
},
)
return False
return True
@@ -47,7 +60,14 @@ def git_pull(repository: str, env: dict[str, str] | None = None) -> bool:
git_env = os.environ | (env or {})
try:
subprocess.check_call(["git", "pull"], cwd=repository, env=git_env)
except subprocess.CalledProcessError:
print(f"Unable to pull repository {repository}")
except subprocess.CalledProcessError as err:
logger.error(
"git pull failed",
extra={
OPERATION: "pull",
PATH: repository,
EXIT_CODE: err.returncode,
},
)
return False
return True
src/gitea_api.py
+18 -1
View File
@@ -1,11 +1,14 @@
import logging
from urllib.parse import urljoin
import requests
from pydantic import TypeAdapter
from .log_fields import OPERATION, PAGE, REPOSITORY_COUNT, STATUS_CODE
from .models import GiteaRepository
REPOSITORIES_ADAPTER = TypeAdapter(list[GiteaRepository])
logger = logging.getLogger(__name__)
class GiteaApi:
@@ -30,7 +33,14 @@ class GiteaApi:
params={"limit": page_size, "page": page_id},
)
if r.status_code != 200:
print(f"Failed request, code {r.status_code}")
logger.error(
"failed to fetch repositories",
extra={
OPERATION: "fetch_repositories",
STATUS_CODE: r.status_code,
PAGE: page_id,
},
)
return []
repos_data = r.json()
if not repos_data:
@@ -40,4 +50,11 @@ class GiteaApi:
cur_repos = REPOSITORIES_ADAPTER.validate_python(repos_data)
for repo in cur_repos:
all_repos[repo.repo_id] = repo
logger.info(
"repositories fetched",
extra={
OPERATION: "fetch_repositories",
REPOSITORY_COUNT: len(all_repos),
},
)
return list(all_repos.values())
src/log_fields.py
+8
View File
@@ -0,0 +1,8 @@
OPERATION = "operation"
REPOSITORY = "repository"
PATH = "path"
REPOSITORY_URL = "repository_url"
EXIT_CODE = "exit_code"
STATUS_CODE = "status_code"
PAGE = "page"
REPOSITORY_COUNT = "repository_count"
src/logging_config.py
+82
View File
@@ -0,0 +1,82 @@
import json
import logging
import logging.handlers
import sys
from datetime import UTC, datetime
from typing import Any
from .models import LoggingConfig, LoggingOutputConfig
RESERVED_LOG_RECORD_KEYS = set(logging.makeLogRecord({}).__dict__)
def _context(record: logging.LogRecord) -> dict[str, Any]:
return {
key: value
for key, value in record.__dict__.items()
if key not in RESERVED_LOG_RECORD_KEYS and not key.startswith("_")
}
class JsonFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
data = {
"timestamp": datetime.fromtimestamp(
record.created,
tz=UTC,
).isoformat(),
"level": record.levelname,
"logger": record.name,
"message": record.getMessage(),
}
data.update(_context(record))
if record.exc_info:
data["exception"] = self.formatException(record.exc_info)
return json.dumps(data, ensure_ascii=False, default=str)
class PlainFormatter(logging.Formatter):
def format(self, record: logging.LogRecord) -> str:
message = f"{record.levelname} {record.getMessage()}"
context = _context(record)
if context:
fields = " ".join(
f"{key}={value}" for key, value in context.items()
)
message = f"{message} {fields}"
if record.exc_info:
message = f"{message}\n{self.formatException(record.exc_info)}"
return message
def _formatter(output: LoggingOutputConfig) -> logging.Formatter:
if output.format == "json":
return JsonFormatter()
return PlainFormatter()
def _handler(output: LoggingOutputConfig) -> logging.Handler:
if output.type == "file":
handler = logging.handlers.RotatingFileHandler(
filename=output.path,
maxBytes=output.max_size,
backupCount=output.backup_count,
)
else:
stream = sys.stderr if output.stream == "stderr" else sys.stdout
handler = logging.StreamHandler(stream)
handler.setFormatter(_formatter(output))
return handler
def setup_logging(
config: LoggingConfig,
logger: logging.Logger | None = None,
) -> logging.Logger:
configured_logger = logger or logging.getLogger()
configured_logger.handlers.clear()
configured_logger.setLevel(config.level)
configured_logger.propagate = False
for output in config.outputs:
configured_logger.addHandler(_handler(output))
return configured_logger
src/models.py
+24
View File
@@ -1,4 +1,5 @@
import datetime
from typing import Literal
from pydantic import BaseModel, Field, HttpUrl, model_validator
@@ -26,6 +27,29 @@ class Config(BaseModel):
api_token: str
out_dir: str
fetch: FetchConfig
logging: "LoggingConfig" = Field(default_factory=lambda: LoggingConfig())
class LoggingOutputConfig(BaseModel):
type: Literal["console", "file"] = "console"
format: Literal["plain", "json"] = "plain"
stream: Literal["stdout", "stderr"] = "stdout"
path: str | None = None
max_size: int = 10 * 1024 * 1024
backup_count: int = 5
@model_validator(mode="after")
def validate_output(self):
if self.type == "file" and not self.path:
raise ValueError("file logging output requires path")
return self
class LoggingConfig(BaseModel):
level: str = "INFO"
outputs: list[LoggingOutputConfig] = Field(
default_factory=lambda: [LoggingOutputConfig()]
)
class GiteaUser(BaseModel):
tests/test_config.py
+67 -1
View File
@@ -2,7 +2,13 @@ from tempfile import NamedTemporaryFile
import pytest
from src.config import Config, FetchConfig, read_toml_config
from src.config import (
Config,
FetchConfig,
LoggingConfig,
read_toml_config,
)
from src.models import LoggingOutputConfig
@pytest.mark.parametrize(
@@ -22,6 +28,7 @@ from src.config import Config, FetchConfig, read_toml_config
out_dir="/home/user/repositories",
endpoint="https://example.com",
fetch=FetchConfig(ssh_key="/tmp/no_key"),
logging=LoggingConfig(),
),
),
(
@@ -39,6 +46,65 @@ from src.config import Config, FetchConfig, read_toml_config
out_dir="/home/user/repositories",
endpoint="https://example.com",
fetch=FetchConfig(user="alex", token="repo-token"),
logging=LoggingConfig(),
),
),
(
'[main]\napi_token = "something"\n'
'format = "{owner}/{name}"\n'
'endpoint = "https://example.com"\n'
'out_dir = "/home/user/repositories"\n'
"\n"
"[fetch]\n"
'ssh_key = "/tmp/no_key"\n'
"\n"
"[logging]\n"
'level = "DEBUG"\n'
"\n"
"[[logging.outputs]]\n"
'type = "console"\n'
'format = "plain"\n'
'stream = "stdout"\n'
"\n"
"[[logging.outputs]]\n"
'type = "console"\n'
'format = "json"\n'
'stream = "stderr"\n'
"\n"
"[[logging.outputs]]\n"
'type = "file"\n'
'path = "/tmp/gitea-mirror.log"\n'
'format = "json"\n'
"max_size = 1048576\n"
"backup_count = 3",
Config(
api_token="something",
repository_format="{owner}/{name}",
out_dir="/home/user/repositories",
endpoint="https://example.com",
fetch=FetchConfig(ssh_key="/tmp/no_key"),
logging=LoggingConfig(
level="DEBUG",
outputs=[
LoggingOutputConfig(
type="console",
format="plain",
stream="stdout",
),
LoggingOutputConfig(
type="console",
format="json",
stream="stderr",
),
LoggingOutputConfig(
type="file",
path="/tmp/gitea-mirror.log",
format="json",
max_size=1048576,
backup_count=3,
),
],
),
),
),
("[main]", None),
tests/test_logging_config.py
+112
View File
@@ -0,0 +1,112 @@
import io
import json
import logging
import logging.handlers
import sys
from src.log_fields import EXIT_CODE, OPERATION, PATH, REPOSITORY
from src.logging_config import JsonFormatter, PlainFormatter, setup_logging
from src.models import LoggingConfig, LoggingOutputConfig
def test_json_formatter_keeps_context_fields():
record = logging.LogRecord(
name="gitea_mirror.git",
level=logging.ERROR,
pathname=__file__,
lineno=10,
msg="git command failed",
args=(),
exc_info=None,
)
setattr(record, OPERATION, "pull")
setattr(record, REPOSITORY, "alex/repo")
setattr(record, PATH, "/backup/alex/repo")
setattr(record, EXIT_CODE, 128)
data = json.loads(JsonFormatter().format(record))
assert data["level"] == "ERROR"
assert data["logger"] == "gitea_mirror.git"
assert data["message"] == "git command failed"
assert data[OPERATION] == "pull"
assert data[REPOSITORY] == "alex/repo"
assert data[PATH] == "/backup/alex/repo"
assert data[EXIT_CODE] == 128
assert "timestamp" in data
def test_plain_formatter_appends_context_fields():
record = logging.LogRecord(
name="gitea_mirror.git",
level=logging.INFO,
pathname=__file__,
lineno=10,
msg="repository cloned",
args=(),
exc_info=None,
)
setattr(record, OPERATION, "clone")
setattr(record, REPOSITORY, "alex/repo")
message = PlainFormatter().format(record)
assert message == (
f"INFO repository cloned {OPERATION}=clone {REPOSITORY}=alex/repo"
)
def test_setup_logging_configures_console_and_rotating_file(tmp_path):
log_path = tmp_path / "gitea-mirror.log"
logger = logging.getLogger("test_setup_logging")
setup_logging(
LoggingConfig(
level="DEBUG",
outputs=[
LoggingOutputConfig(
type="console",
format="plain",
stream="stdout",
),
LoggingOutputConfig(
type="console",
format="json",
stream="stderr",
),
LoggingOutputConfig(
type="file",
path=str(log_path),
format="json",
max_size=1024,
backup_count=2,
),
],
),
logger=logger,
)
assert logger.level == logging.DEBUG
assert len(logger.handlers) == 3
assert logger.propagate is False
stdout_handler, stderr_handler, file_handler = logger.handlers
assert stdout_handler.stream is sys.stdout
assert isinstance(stdout_handler.formatter, PlainFormatter)
assert stderr_handler.stream is sys.stderr
assert isinstance(stderr_handler.formatter, JsonFormatter)
assert isinstance(file_handler, logging.handlers.RotatingFileHandler)
assert isinstance(file_handler.formatter, JsonFormatter)
assert file_handler.maxBytes == 1024
assert file_handler.backupCount == 2
def test_setup_logging_replaces_existing_handlers():
logger = logging.getLogger("test_setup_logging_replaces")
old_stream = io.StringIO()
logger.addHandler(logging.StreamHandler(old_stream))
setup_logging(LoggingConfig(), logger=logger)
assert len(logger.handlers) == 1
assert isinstance(logger.handlers[0].formatter, PlainFormatter)