jonashaag

import textwrap
from dataclasses import dataclass


@dataclass
class Pattern:
    pattern: str
    prematchers: list[str]

jonashaag

#!/bin/bash

set -euo pipefail

if [ $# -lt 2 ]; then
  echo "Usage: $0 SCHEDULE PROG [ARGS]..." >&2
  echo "SCHEDULE is used in 'date -d <SCHEDULE>'." >&2
  echo "Example: $0 '1 hour' myprog --arg" >&2
  exit 1
fi

jonashaag

def run_test_in_fork(func, *args, **kwargs):
    """Run a pytest test in a fork of the pytest process.

    Useful to check behaviour of some code when run in a forked process.
    The test outcome will be reported normally in the pytest parent.
    """
    if not hasattr(os, "fork"):
        pytest.skip("os.fork not available")
    error_in_child = multiprocessing.Value("b")
    child_pid = os.fork()

jonashaag

function __micromamba_completion
    set -l args (commandline -ocp)
    # TODO: micromamba completer should be able to ignore micromamba program name (first arg)
    set -e args[1]
    set -l suggestions ($MAMBA_EXE completer $args (commandline -t))
    # TODO: micromamba completer should output lines not columnified
    if echo "$suggestions" | grep -q "    "
      set suggestions (string split " " "$suggestions")
    end
    for suggestion in $suggestions

jonashaag

UPDATE pg_index
SET indisready=false, indisvalid=false
WHERE indrelid = (SELECT oid FROM pg_class WHERE relname='<TABLE_NAME>');

... do work ...

REINDEX TABLE "<TABLE_NAME>";

jonashaag

import shutil
from pathlib import Path


def s3_copytree(bucket, src_prefix: str, dst_folder: Path):
    for s3_f in bucket.objects.filter(Prefix=src_prefix):
        target_path = dst_folder / s3_f.key.removeprefix(src_prefix)
        target_path.parent.mkdir(parents=True, exist_ok=True)
        with open(target_path, "wb") as target_f:
            shutil.copyfileobj(s3_f.get()["Body"], target_f)

jonashaag

declare const KV: KVNamespace

addEventListener('fetch', event => event.respondWith(handleRequest(event.request)))

const handleRequest = async (request: Request): Promise<Response> => {
  if (!auth(request)) {
    return new Response('Forbidden', { status: 403 })
  }

  const key = new URL(request.url).pathname.substr(1)

jonashaag

from cython cimport Py_ssize_t
from libc.stdint cimport (
    uint8_t,
    uint16_t,
    uint32_t,
    uint64_t,
)


def read_float_with_byteswap(bytes data, Py_ssize_t offset, bint byteswap):

jonashaag

import atexit
import base64
import logging
import os
import pickle

import diskcache

import proxy2

jonashaag

def pack(arr):
    """Pack integers < 2**12 into 12 bit integers, encoded as bytes"""
    out = []
    for i in range(0, len(arr), 2):
        e1, e2 = arr[i:i+2]
        assert 0 <= e1 < 2**12
        assert 0 <= e2 < 2**12
        e1 = (e1 << 4) | (e2 >> 8)
        e2 &= 2**8-1
        assert e1 <= 2**16-1