antonagestam’s gists

antonagestam / multimodel.py

Last active October 5, 2024 11:40

Pydantic partial/multi models

	from collections.abc import Mapping
	from typing import Any

	from pydantic import BaseModel
	from pydantic import root_validator
	from pydantic.fields import ModelField


	class PartialModel(BaseModel): ...

antonagestam / eager-load.py

Created March 3, 2024 11:29

Eagerly load all kio modules, takes 2-3s

	import kio.schema
	from pkgutil import iter_modules
	from pkgutil import resolve_name
	import time
	from pathlib import Path

	schema_dir = Path(kio.schema.__file__).parent.resolve()
	print(f"{schema_dir=}")

	t0 = time.monotonic_ns()

antonagestam / join_streams.py

Last active April 5, 2025 03:48

Merge results from multiple async generators into one single stream.

	import asyncio
	import random
	from typing import TypeVar, AsyncGenerator

	T = TypeVar("T")


	async def read_into_queue(
	task: AsyncGenerator[T, None],
	queue: asyncio.Queue[T],

antonagestam / kafka-meta.schema.json

Last active February 12, 2024 20:28

A JSONSchema to describe the shape of Apache Kafka's protocol message schemas.

	{
	"$schema": "https://json-schema.org/draft/2020-12/schema",
	"type": "object",
	"additionalProperties": false,
	"oneOf": [
	{
	"properties": {
	"apiKey": {"type": "number"},
	"type": {
	"type": "string",

antonagestam / apply_factory.py

Created January 28, 2022 23:34

	from typing import Callable
	from typing import TypeVar

	from phantom.predicates.generic import equal
	from phantom import Predicate

	T = TypeVar("T")


	def apply_factory(factory: Callable[[T], Predicate[T]]) -> Predicate[tuple[T, T]]:

antonagestam / rss-siginfo.py

Created June 18, 2021 09:36

	import time
	import signal
	import resource


	def debug(signum, frame):
	print(f"RSS: {resource.getrusage(resource.RUSAGE_SELF).ru_maxrss}")
	print(f"{resource.getpagesize()=}")

antonagestam / phantom-literal.py

Created June 9, 2021 08:47

Recipe for a powerful combination of Literals and phantom types ...

	from phantom import Phantom
	from typing import Literal, get_args, Union, Mapping

	from phantom.predicates.collection import contained

	LiteralCountry = Literal["SE", "DE", "DK"]


	class PhantomCountry(str, Phantom, predicate=contained(get_args(LiteralCountry))):
	...

antonagestam / trunk-based-release-flow.md

Created February 20, 2021 12:25

Handling trunk-based releases with issues and tags

Handling trunk-based release approval with issues and tags

Command to create a release.

make request-release <sha>
make request-release  # default to current trunk

antonagestam / composable.py

Last active February 12, 2021 13:12

Functional composition with cute syntax + typing support.

	from __future__ import annotations
	from typing import TypeVar, Generic, Callable

	Arg = TypeVar("Arg")
	Ret = TypeVar("Ret")
	NewRet = TypeVar("NewRet")


	class Composable(Generic[Arg, Ret]):
	def __init__(self, fn: Callable[[Arg], Ret]) -> None:

antonagestam / choice-constraints-wip.py

Last active May 20, 2021 09:16

Introspective choice constraints for Django models.

	# Note: This doesn't properly yield a delete+create when choices change.
	# -> This can probably be remedied by requiring passing the model "path" (<app>.<model_name>) as
	# an argument to __init__. Not ideal ...
	from typing import Callable, Sequence

	from django.db import models
	from django.db.backends.base.schema import BaseDatabaseSchemaEditor
	from django.db.models.constraints import BaseConstraint
	from django.db.models.sql.query import Query