antonagestam

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

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