antonagestam

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

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

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

antonagestam

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

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

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

Handling trunk-based release approval with issues and tags

Requesting a release

Command to create a release.

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

antonagestam

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

# 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

antonagestam

from __future__ import annotations

import dataclasses
from typing import Annotated
from typing import Callable
from typing import Generic
from typing import Iterable
from typing import Mapping
from typing import Tuple
from typing import Type