corekit

Shared foundations for Python projects: structured logging, benchmarking, registries, FastAPI routers with built-in handlers, an in-memory record store, and ETL scaffolding.

Requires Python 3.11+.

Install

pip install python-corekit

The distribution is python-corekit; the import is corekit.

Every dependency corekit needs is installed with it. There are no optional extras to remember, and no import that fails because something was left out.

Pin a compatible release rather than tracking whatever is newest:

python-corekit~=0.1.0

Before 1.0, the minor version carries breaking changes.

Logging

Inherit from Loggable and every instance gets a logger named after its class.

from corekit.observability import Loggable

class Importer(Loggable):
    def run(self) -> None:
        self.info("starting")
        try:
            ...
        except Exception:
            self.exception("import failed", exc_info=True)

Benchmarkable adds split timing on top:

from corekit.observability import Benchmarkable

class Report(Benchmarkable):
    def build(self) -> None:
        self.timing()            # start the clock
        ...
        self.timing("queried")   # logs the time since the previous split

Routers and handlers

A router and the handler holding its business logic travel together. Declare the handler type in square brackets and the router builds it for you.

from corekit.api import BaseHandler, SmartRouter

class AdminHandler(BaseHandler):
    """
    Admin operations.

    Handlers inherit logging and benchmarking, and register themselves by name.
    """

    async def list_users(self) -> list[str]:
        return ["ada", "bob"]

router = SmartRouter[AdminHandler](route_prefix="/admin", tags=["Admin"])

@router.get("/users")
async def list_users() -> list[str]:
    return await router.handler.list_users()

Mount it with router.include(app) — the router adds itself, rather than the application having to know about it.

The handler is built on first use, and router.handler can be assigned, so tests can substitute a double without constructing the real thing:

router.handler = FakeAdminHandler()

Handlers register themselves under a normalized name, so any spelling finds them:

BaseHandler.get_handler_by_name("admin_handler")   # also "AdminHandler", "Admin Handler"

Datasets

An in-memory, schema-fixed collection with composable filters. Standard library only — no pandas.

from corekit.data import Dataset, Field

people = Dataset(id_key="name", schema=["name", "age"])
people.add({"name": "Ada", "age": 36})
people.add({"name": "Bob", "age": 17})

adults = people.filter(Field("age") >= 18)
people.get_record("Ada").age            # O(1) lookup by id

Stores pickle cleanly, including their dynamically generated record class.

Homelab pieces

Container control

from corekit.docker import Watchdog

watchdog = Watchdog(enforce_label=True)
watchdog.restart_container_by_name("minecraft")
watchdog.find_and_stop(label="app", value="staging")

enforce_label limits the blast radius: with it on, only containers carrying the watchdog=true label can be started, stopped or paused, so a mistyped name cannot take down something unrelated. Leave it on unless the watchdog is meant to control everything on the host.

Reacting to logs

Describe what to watch for and what to do about it:

# config.yaml
containers:
  - name: "minecraft-.*"
    rules:
      - name: "out of memory"
        pattern: "java.lang.OutOfMemoryError"
        severity: critical
        send_notification: true
        actions:
          - type: restart_container
            max_restarts: 3
            restart_window: 3600
advanced:
  ignore_patterns:
    - "healthcheck"
  rate_limits:
    restart_container:
      count: 5
      period: hour

from corekit.log_monitor import LogMonitor

LogMonitor.run("config.yaml")

Restarts are capped per container, so a crash loop cannot become a restart loop.

Notifications

from corekit.notifications import BaseNotificationService, Notification, NotificationType

class DiscordNotifier(BaseNotificationService):
    """
    Sends notifications to a Discord channel.
    """

    def _send(self, message: str) -> None:
        discord.post(message)

notifier.notify(Notification(message="disk full", type=NotificationType.ERROR))

Override _send, not send. notify() formats the message and calls _send, so an override with any other name is silently ignored.

Real-time updates

Publish from wherever the work happens:

from corekit.events import EventPublisher

publisher = EventPublisher.for_resource("minecraft", "server", "survival")
publisher.publish("backup_finished", {"size": "4.2GB"})

Stream it to the browser:

from corekit.api import SSEResponse
from corekit.events import SSEStream

@router.get("/events")
async def events(channel: str) -> SSEResponse:
    return SSEResponse(SSEStream(channel, keepalive_interval=15))

The browser side is three lines, and reconnects on its own:

const source = new EventSource("/events?channel=minecraft:server:survival");
source.addEventListener("backup_finished", e => console.log(JSON.parse(e.data)));

SSEStream sends a connected frame on subscribe, an optional initial_state so a client arriving late renders immediately, and a comment frame every keepalive_interval seconds so proxies do not close an idle connection. For WebSockets, WebSocketBridge relays the same channel and stops on a terminal status.

Publishing never raises: an event that cannot be delivered should not take down the operation that produced it. publish returns whether it worked.

Parallel work

from corekit.concurrency import parallelize

@parallelize()
def fetch(url: str) -> Response:
    return client.get(url)

for response in fetch(urls):
    ...

Results arrive as they finish; pass ordered=True for input order. The thread count comes from concurrency.default_threads unless you name one, and is capped at max_threads either way — asking for 9,999 threads gets you the ceiling, not 9,999 threads.

Failures propagate by default. Pass raise_on_error=False to log and skip them instead, which loses results silently and so is opt-in.

HTTP clients

from corekit.http.client import BaseApiClient

class GithubClient(BaseApiClient):
    """
    Talks to the GitHub API.
    """

    @property
    def base_url(self) -> str:
        return "https://api.github.com"

response = GithubClient().get("/users/octocat")
response.data["login"]

Retries 429 and 5xx with exponential backoff. Every response is a BaseApiResponse, so a non-JSON error page leaves data empty rather than raising. async_get, async_post and friends do the same without blocking.

Serialization

from corekit.serialization.serializer import Serializer
from corekit.serialization.enum import SerializerEngine

serializer = Serializer(SerializerEngine.JSON)
serializer.deserialize(serializer.serialize({"a": 1}))

JSON is the default because it cannot execute code. pickle and dill can, so selecting either requires a key, and payloads are authenticated with an HMAC that is verified before anything is decoded:

Serializer(SerializerEngine.PICKLE, key=os.environ["APP_KEY"])

Never deserialize untrusted bytes with an engine that executes code, even signed. The key proves the payload came from you, not that its contents are safe.

Configuration

Configuration is optional. corekit never reads the environment at import time, so importing it can never fail for want of a variable.

Precedence, highest first: explicit argument, environment, config file, default.

# corekit.toml, or a [tool.corekit] table in pyproject.toml
[standards]
require_handler_docstrings = true

[concurrency]
default_threads = 4      # used when a caller does not say
max_threads = 32         # never exceeded, however it is asked

[database]
url = "postgresql://localhost/app"

[crypto]
salt = "..."

Settings are grouped by concern, so get_settings().concurrency.max_threads says where a value belongs. Environment variables use a double underscore for the section: COREKIT_CONCURRENCY__MAX_THREADS=16.

Environment variables use a COREKIT_ prefix (COREKIT_CRYPTO_SALT). Empty values are treated as unset, because container runtimes routinely pass FOO= for a variable that was never set.

from corekit.config import CorekitSettings, StandardsSettings, set_settings

set_settings(CorekitSettings(standards=StandardsSettings(require_handler_docstrings=True)))

Requiring docstrings

Off by default. Turn it on and every BaseHandler subclass must carry a multiline docstring or fail at import. Individual classes can opt out with __require_doc__ = False.

Layout

Packages are named for what they are, and sit in the layer they belong to. Imports go downward only.

corekit/
  config/                          settings, sources, loader
  constants.py

  exceptions/                      error types

  observability/                   Loggable, Benchmarkable, Timer
  registry/  schemas/  utils/      SmartRegistry, enums and fields, helpers
  data/                            Dataset and its filter expressions
  crypto/  files/  serialization/
  concurrency/                     ThreadLocalRegistry, ThreadWorker
  decorators/

  connections/                     the Connectable lifecycle and @connect
    sql/                           SQLConnection, queries, migrations
    redis/                         RedisConnection
  http/                            BaseApiClient, retries, responses

  api/                             handlers, routers, responses
  docker/  notifications/  etl/

  events/  log_monitor/            built on the capabilities above

sql and redis sit under connections because both implement Connectable. docker does not -- Watchdog manages containers and has no connection lifecycle -- so it stays a top-level integration.

tests/test_architecture.py enforces the direction: it fails on a cycle, on an import pointing upward, or on a new package that has not been placed in the layering deliberately.

Development

pip install -e ".[dev]"
pytest
ruff format . && ruff check --fix .

tests/test_imports.py imports every module in the package, so a module that nothing else happens to import still has to be importable.

Licence

MIT.