jonashaag

def unload_to_parquet(query: str, target_dir: Path, conn, stage_name: str = "unload_stage"):
    conn.execute(f"CREATE TEMP STAGE {stage_name}")
    conn.execute(f"COPY INTO @{stage_Name} FROM ({query}) file_format=(type='parquet') header=true")
    target_dir.mkdir(parents=True)
    conn.execute(f"GET @{stage_name} file://{str(target_dir)}")

jonashaag

import numpy as np
import pandas as pd
from pandas.api.types import is_numeric_dtype
from pandas.core.dtypes.base import ExtensionDtype


def shrink_dtype(series: pd.Series) -> pd.Series:
    smallest_dtype = get_smallest_dtype(series)
    if smallest_dtype == series.dtype:
        return series

jonashaag

import json
import sqlite3

repodata = json.load(open("497deca9.json"))

COLS = 'filename, build, build_number, depends, license, license_family, md5, name, sha256, size, subdir, timestamp, version'.split(', ')

db = sqlite3.connect("497deca9.sqlite")
db.execute("create table repodata ({}, primary key (filename))".format(','.join(COLS)))

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)