trixy_core.utils

Utility-Module mit Logging, Decorators, Validation und Rate-Limiting.

Übersicht

from trixy_core.utils import (
    # Debug-Funktionen
    pinfo, pdebug, perror, pwarn, set_debug_mode,
    # Version
    VERSION, VERSION_STRING, PROTOCOL_VERSION,
)

Debug-Funktionen

pinfo(message, *args)

Gibt eine Info-Nachricht aus.

from trixy_core.utils import pinfo

pinfo("Server gestartet auf Port %d", 2101)

pdebug(message, *args)

Gibt eine Debug-Nachricht aus (nur wenn Debug-Modus aktiv).

from trixy_core.utils import pdebug

pdebug("Empfangene Daten: %s", data)

perror(message, *args)

Gibt eine Fehlermeldung aus.

from trixy_core.utils import perror

perror("Verbindung fehlgeschlagen: %s", error)

pwarn(message, *args)

Gibt eine Warnung aus.

from trixy_core.utils import pwarn

pwarn("Konfiguration fehlt, verwende Standardwerte")

set_debug_mode(enabled: bool)

Aktiviert/deaktiviert den Debug-Modus.

from trixy_core.utils import set_debug_mode

set_debug_mode(True)  # Debug-Ausgaben aktivieren

trixy_core.utils.logging

Strukturiertes Logging mit JSON-Unterstützung und Correlation-IDs.

Import

from trixy_core.utils.logging import (
    StructuredLogger,
    get_logger,
    LogContext,
    get_correlation_id,
    set_correlation_id,
    JSONFormatter,
    TextFormatter,
)

StructuredLogger

Logger mit strukturierter Ausgabe und Kontext-Propagation.

from trixy_core.utils.logging import StructuredLogger

logger = StructuredLogger("my_module")

# Einfaches Logging
logger.info("Benutzer angemeldet", user_id=123, ip="192.168.1.1")

# Mit Kontext
with logger.context(request_id="abc123"):
    logger.debug("Verarbeite Anfrage")
    logger.info("Anfrage abgeschlossen", duration=0.5)

# Fehler mit Traceback
try:
    process()
except Exception as e:
    logger.exception("Verarbeitung fehlgeschlagen", error=str(e))

Methoden:

Methode	Beschreibung
`debug(msg, **kwargs)`	Debug-Level
`info(msg, **kwargs)`	Info-Level
`warning(msg, **kwargs)`	Warning-Level
`error(msg, **kwargs)`	Error-Level
`exception(msg, **kwargs)`	Error mit Traceback
`context(**kwargs)`	Context-Manager für zusätzliche Felder

LogContext

Kontext für Request-Tracing.

from trixy_core.utils.logging import LogContext, set_correlation_id

# Correlation-ID setzen
set_correlation_id("req-12345")

# Kontext für alle nachfolgenden Logs
with LogContext(user_id=123, session="abc"):
    logger.info("Operation gestartet")

Formatter

from trixy_core.utils.logging import JSONFormatter, TextFormatter, CompactFormatter

# JSON für Produktion
json_formatter = JSONFormatter(include_timestamp=True)

# Text für Entwicklung
text_formatter = TextFormatter(colorize=True)

# Kompakt für Logs
compact_formatter = CompactFormatter()

trixy_core.utils.decorators

Wiederverwendbare Decorators für Timing, Caching, Retry und Validation.

Import

from trixy_core.utils.decorators import (
    timed, async_timed, profile,
    cached, async_cached, cache_clear,
    retry, async_retry,
    deprecated, pending_deprecation,
    validate, validate_args, validate_return,
)

@timed

Misst und loggt die Ausführungszeit.

from trixy_core.utils.decorators import timed, async_timed

@timed(name="db_query", threshold=0.5)
def query_database(sql: str) -> list:
    return db.execute(sql)

@async_timed(name="api_call")
async def fetch_api(url: str) -> dict:
    async with session.get(url) as response:
        return await response.json()

Parameter:

Parameter	Typ	Default	Beschreibung
`name`	`str`	Funktionsname	Name für Logs
`threshold`	`float`	`None`	Warnung wenn überschritten
`logger`	`Logger`	`None`	Benutzerdefinierter Logger

@cached

LRU-Cache mit optionalem TTL.

from trixy_core.utils.decorators import cached, async_cached, cache_clear

@cached(ttl=300, max_size=100)
def get_user(user_id: int) -> User:
    return db.query(User).get(user_id)

@async_cached(ttl=60)
async def fetch_config(key: str) -> dict:
    return await redis.get(key)

# Cache leeren
cache_clear(get_user)

Parameter:

Parameter	Typ	Default	Beschreibung
`ttl`	`float`	`None`	Time-to-live in Sekunden
`max_size`	`int`	`128`	Maximale Cache-Größe
`key_func`	`Callable`	`None`	Benutzerdefinierte Key-Funktion

@retry

Wiederholt bei Fehlern mit Backoff.

from trixy_core.utils.decorators import retry, async_retry, RetryConfig

@retry(max_attempts=3, backoff=1.0, exceptions=(ConnectionError, TimeoutError))
def connect_to_server(host: str) -> Connection:
    return socket.connect(host)

@async_retry(max_attempts=5, backoff=2.0, jitter=0.5)
async def send_request(data: dict) -> Response:
    return await client.post("/api", json=data)

# Mit Konfiguration
config = RetryConfig(
    max_attempts=3,
    backoff=1.0,
    max_backoff=30.0,
    jitter=0.1,
    exceptions=(IOError,),
)

@retry(config=config)
def read_file(path: str) -> str:
    return Path(path).read_text()

Parameter:

Parameter	Typ	Default	Beschreibung
`max_attempts`	`int`	`3`	Maximale Versuche
`backoff`	`float`	`1.0`	Basis-Wartezeit
`max_backoff`	`float`	`60.0`	Maximale Wartezeit
`jitter`	`float`	`0.0`	Zufällige Variation
`exceptions`	`tuple`	`(Exception,)`	Zu fangene Exceptions

@deprecated

Markiert Funktionen als veraltet.

from trixy_core.utils.decorators import deprecated, pending_deprecation

@deprecated(message="Verwende new_function() stattdessen", version="2.0.0")
def old_function():
    pass

@pending_deprecation(message="Wird in 3.0.0 entfernt")
def legacy_function():
    pass

@validate

Validiert Funktionsparameter.

from trixy_core.utils.decorators import validate, validate_args, validate_return

@validate(
    name=str,
    age=(int, lambda x: x >= 0),
    email=(str, lambda x: "@" in x),
)
def create_user(name: str, age: int, email: str) -> User:
    return User(name=name, age=age, email=email)

@validate_args(
    url=(str, lambda x: x.startswith("http")),
)
@validate_return(dict)
async def fetch_json(url: str) -> dict:
    return await client.get(url).json()

trixy_core.utils.ratelimit

Token-Bucket und Sliding-Window Rate-Limiting.

Import

from trixy_core.utils.ratelimit import (
    TokenBucket,
    SlidingWindow,
    rate_limit,
    async_rate_limit,
    RateLimitExceeded,
    RateLimitConfig,
)

TokenBucket

Token-Bucket-Algorithmus für Rate-Limiting.

from trixy_core.utils.ratelimit import TokenBucket

# 10 Requests pro Sekunde
bucket = TokenBucket(rate=10, capacity=10)

if bucket.consume():
    process_request()
else:
    raise RateLimitExceeded("Zu viele Anfragen")

# Mehrere Tokens auf einmal
if bucket.consume(tokens=5):
    process_batch()

Parameter:

Parameter	Typ	Beschreibung
`rate`	`float`	Tokens pro Sekunde
`capacity`	`int`	Maximale Token-Anzahl

SlidingWindow

Sliding-Window-Algorithmus für präziseres Limiting.

from trixy_core.utils.ratelimit import SlidingWindow

# 100 Requests pro Minute
window = SlidingWindow(limit=100, window_size=60)

if window.allow():
    process_request()
else:
    retry_after = window.retry_after()
    raise RateLimitExceeded(f"Retry nach {retry_after}s")

@rate_limit Decorator

from trixy_core.utils.ratelimit import rate_limit, async_rate_limit

@rate_limit(rate=10, capacity=10)
def api_endpoint(request):
    return process(request)

@async_rate_limit(limit=100, window_size=60, strategy="sliding_window")
async def async_endpoint(request):
    return await process(request)

trixy_core.utils.validation

Schema-basierte Validierung für Dictionaries und Dataclasses.

Import

from trixy_core.utils.validation import (
    # Validators
    TypeValidator, RangeValidator, PatternValidator,
    LengthValidator, RequiredValidator, EnumValidator,
    ListValidator, DictValidator, CompositeValidator,
    # Schema
    Schema, Field, NestedField, ListField,
    # Functions
    validate_dict, validate_dataclass,
    # Exceptions
    ValidationError, ValidationErrors,
)

Validators

from trixy_core.utils.validation import (
    TypeValidator, RangeValidator, PatternValidator,
    LengthValidator, CompositeValidator,
)

# Typ-Validierung
type_v = TypeValidator(int)
type_v.validate(42)  # OK
type_v.validate("42")  # ValidationError

# Bereichs-Validierung
range_v = RangeValidator(min_value=0, max_value=100)
range_v.validate(50)  # OK

# Pattern-Validierung
pattern_v = PatternValidator(r"^[a-z]+@[a-z]+\.[a-z]+$")
pattern_v.validate("user@example.com")  # OK

# Längen-Validierung
length_v = LengthValidator(min_length=3, max_length=50)
length_v.validate("hello")  # OK

# Kombinierte Validierung
composite = CompositeValidator([
    TypeValidator(str),
    LengthValidator(min_length=1),
    PatternValidator(r"^[A-Za-z]+$"),
])
composite.validate("Hello")  # OK

Schema

from trixy_core.utils.validation import Schema, Field, NestedField, ListField

class UserSchema(Schema):
    name = Field(str, required=True, min_length=1, max_length=100)
    age = Field(int, required=True, min_value=0, max_value=150)
    email = Field(str, required=True, pattern=r"^.+@.+\..+$")
    roles = ListField(str, min_items=1)
    address = NestedField(AddressSchema, required=False)

# Validieren
schema = UserSchema()
errors = schema.validate({
    "name": "Max",
    "age": 25,
    "email": "max@example.com",
    "roles": ["user"],
})

if errors:
    for error in errors:
        print(f"{error.field}: {error.message}")

validate_dict

from trixy_core.utils.validation import validate_dict

schema = {
    "name": {"type": str, "required": True},
    "age": {"type": int, "min": 0},
    "tags": {"type": list, "items": str},
}

data = {"name": "Test", "age": 25, "tags": ["a", "b"]}
errors = validate_dict(data, schema)

utils.md 10 KB History Raw

trixy_core.utils

Übersicht

Debug-Funktionen

pinfo(message, *args)

pdebug(message, *args)

perror(message, *args)

pwarn(message, *args)

set_debug_mode(enabled: bool)

trixy_core.utils.logging

Import

StructuredLogger

LogContext

Formatter

trixy_core.utils.decorators

Import

@timed

@cached

@retry

@deprecated

@validate

trixy_core.utils.ratelimit

Import

TokenBucket

SlidingWindow

@rate_limit Decorator

trixy_core.utils.validation

Import

Validators

Schema

validate_dict

utils.md 10 KB

History Raw