corekit

Shared foundations for Python projects: structured logging, benchmarking, registries, a FastAPI application with routers and handlers, SQL statements and migrations, background tasks, 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.3.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

Assembling an application

Application is a plain FastAPI subclass — every constructor argument, including lifespan, passes straight through. What it adds is a short set of assembly steps, each of which logs what it did.

from corekit.api import Application, Lifespan
from app.backend import routers

lifespan = Lifespan()
lifespan.add("cache", startup=cache.connect, shutdown=cache.disconnect)

app = Application(lifespan=lifespan)
app.discover_routers(routers)

discover_routers imports every module under the package so each router can register itself. It raises if it finds none: passing a package is a statement that routers live there, and an app that silently serves nothing is worse than one that refuses to start.

Lifespan steps start in the order added and shut down in reverse, the way nested with blocks unwind. If a step fails on the way up, the steps that already started are still torn down. A shutdown hook runs only if its own startup completed, so it may assume the state that startup builds — and a hook that raises cannot stop the unwind.

Middleware

Middleware order is a security property: a host check that reads a client address before the proxy-header layer has rewritten it is checking the proxy, not the client. So middleware is never auto-discovered, and corekit installs none by default. A stack is declared in one place, outermost first — the order a request actually meets the layers.

from corekit.api import MiddlewareStack

stack = (
    MiddlewareStack()
    .add(ProxyHeadersMiddleware, trusted_hosts="*")
    .add(TrustedHostMiddleware, allowed_hosts=HOSTS)
    .add(CORSMiddleware, allow_origins=ORIGINS, allow_credentials=True)
)
app.add_middleware_stack(stack)

Reading top to bottom gives the order a request travels, which is the property you need when reviewing it.

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.

SQL statements

Select, insert, update and delete are objects you build and then execute, so a statement can be assembled in pieces and passed around before it runs.

from corekit.connections.sql import SQLConnection, Select, Insert, Update, Delete
from corekit.data import Field

conn = SQLConnection("sqlite:///app.db")

conn.execute(Insert(table=User, rows=[{"name": "Ada", "age": 36}]).add(name="Bob", age=17))

adults = conn.fetch(Select(table=User).where(User.age >= 18).order_by(User.name).limit(10))

conn.execute(Update(table=User).where(User.name == "Bob").set(age=18))
conn.execute(Delete(table=User).where(User.age < 13))

where accepts a SQLAlchemy expression or a corekit.data one, so the same predicate language that filters a Dataset also filters a table:

Select(table=User).where(Field("age") >= 18)

Update and delete compile to a single statement rather than fetching rows and looping, which matters most over a network, where fetch-and-loop pays a round trip per row.

A Delete or Update with no condition raises rather than running:

Delete on User needs a condition; use truncate to empty a table

Background tasks

Tasks describe work and register themselves by name. Nothing here imports a queue library, so the same task runs under RQ, Celery, a cron entry, or a test with no queue at all.

from corekit.jobs import Task, run_task
from corekit.utils import encode_payload

class SendDigest(Task):
    def task_function(self, user: str) -> None:
        ...

# a worker, holding only a name and a JSON string
run_task("SendDigest", encode_payload(["ada"]))

Arguments cross the queue as JSON, never as a serialized object. pickle and dill execute code while loading, so a queue holding objects turns write access to the queue into code execution in a worker. encode_payload and decode_payload live in corekit.utils, since crossing a process boundary as data is not specific to queues.

ScheduledTask adds an interval for tasks a scheduler should repeat.

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 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. The same methods are awaitable inside a running event loop: await client.get(...).

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

  exceptions/                      error types

  observability/                   Loggable, Benchmarkable, Timer
  registry/  schemas/  utils/      registries, enums and fields, helpers
  data/                            Dataset and its filter expressions
  jobs/                            queue-independent background tasks
  crypto/  files/  serialization/
  concurrency/                     ThreadLocalRegistry, ThreadWorker
  decorators/

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

  api/                             Application, lifespan, middleware, routers
  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.