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

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

{
    "$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 Dict, Optional, Any

from django.db import models
from django.core.validators import MinValueValidator, MaxValueValidator
from django import forms


class BaseBoundedField(models.Field):
    def __init__(
            self,

antonagestam

cmake_minimum_required(VERSION 3.2)

project(YourProject)

add_subdirectory(src)
add_subdirectory(lib)

antonagestam

$ systemctl enable rqworker@worker0{1..4}.service
$ systemctl start rqworker@worker0{1..4}.service
$ systemctl status 'rqworker@*'
$ journalctl -u 'rqworker@*'

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

# 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