michalc

import asyncio
import contextlib
import weakref

from fifolock import FifoLock

class ReadAncestor(asyncio.Future):
    @staticmethod
    def is_compatible(holds):
        return not holds[Write]

michalc

from ctypes import (
    byref,
    c_char_p, c_uint32, c_uint16, c_ubyte,
    CDLL,
    POINTER,
    pointer,
    Structure, Union,
    sizeof,
)
import platform

michalc

import csv
import logging

import gevent
from psycopg2 import (
    connect,
    sql,
)

from django.conf import (

michalc

{
  "$schema": "https://vega.github.io/schema/vega-lite/v5.json",
  "width": 630,
  "height": 630,
  "data": {
    "values": [
      {
        "type": "Feature",
        "id": 0,
        "geometry": {

michalc

# From https://stackoverflow.com/a/68876046/1319998, which is itself inspired by https://stackoverflow.com/a/68814418/1319998

from contextlib import contextmanager
from collections import namedtuple
from ctypes import cdll, byref, string_at, c_char_p, c_int, c_double, c_int64, c_void_p
from ctypes.util import find_library
from sys import platform


def query(db_file, sql, params=()):

michalc

/*****************************************************************************/

const hawkId = pm.variables.get('hawk_id');
const hawkKey = pm.variables.get('hawk_key');
const hawkHeader = pm.variables.get('hawk_header') || 'authorization';

/*****************************************************************************/

const timestamp = parseInt(new Date().getTime() / 1000);
const nonce = CryptoJS.enc.Base64.stringify(CryptoJS.lib.WordArray.random(6));

michalc

from contextlib import contextmanager
from ctypes import POINTER, cdll, c_char_p, c_void_p, c_int, create_string_buffer, byref
from sys import platform

# Uses a _little_ endian CTR counter, which OpenSSL doesn't directly support.
# Could be used to decrypt AES-encrypted ZIP files
def decrypt_aes_256_ctr_little_endian(
    key, ciphertext_chunks,
    get_libcrypto=lambda: cdll.LoadLibrary({'linux': 'libcrypto.so', 'darwin': 'libcrypto.dylib'}[platform])
):

michalc

class MyPipeline(_PipelineV2):
    # Everything is a method so nothing happens on import time for flexibility (although possibly
    # does a bit of discovery magic... need to think about that...)
    # Everything is a _static_ method: nothing on self since things are run on different bits of hardware,
    # and gets any run-time dependencies injected in
    #
    # _PipelineV2 won't actually have any code: other parts of the system will interrogate its
    # subclasses as needed. For example
    # - Code in Data Flow would construct a DAG
    # - The test harness would the run this and upstream pipelines synchronously

michalc

# Often it's claimed that a Zip 2.0 file cannot be bigger than 4GiB
# Here's how to make one that's just under 8GiB
from datetime import datetime
from stream_zip import stream_zip, ZIP_32

now = datetime.now()
perms = 0o600

def files():
    for i in range(0, 0xffff):

michalc

import base64
import json

import boto3
from cryptography.hazmat.primitives.ciphers.aead import AESGCM

s3 = boto3.resource('s3')
bucket = s3.Bucket('my-bucket')
kms_client = boto3.client('kms')