tekumara

[Unit]
Description=Jupyter Notebook
After=network-online.target

[Service]
Type=simple
ExecStart=/usr/local/bin/jupyter notebook
Environment="LD_LIBRARY_PATH=/usr/local/cuda/lib64:/usr/local/lib:/usr/lib:/usr/local/cuda/extras/CUPTI/lib64"
User=ubuntu
Group=ubuntu

tekumara

$ kill -QUIT $PID 
SIGQUIT: quit
PC=0x45b521 m=0 sigcode=0

goroutine 0 [idle]:
runtime.futex(0x165f8c8, 0x0, 0x0, 0x0, 0x0, 0xc4207eed80, 0x0, 0x0, 0x7ffef4c7e778, 0x40f19b, ...)
	/usr/local/go/src/runtime/sys_linux_amd64.s:530 +0x21
runtime.futexsleep(0x165f8c8, 0x0, 0xffffffffffffffff)
	/usr/local/go/src/runtime/os_linux.go:45 +0x4b

tekumara

import asyncio
import time


async def heartbeat():
    while True:
        start = time.time()
        await asyncio.sleep(1)
        delay = time.time() - start - 1
        print(f"heartbeat delay = {delay:.3f}s")

tekumara

FROM public.ecr.aws/lambda/python:3.8 as builder
RUN yum install -y zip

WORKDIR /var/task/
COPY setup.py pyproject.toml ./
COPY src ./src/
RUN pip install --no-cache-dir . --target dist

WORKDIR /var/task/dist
RUN zip --quiet -r9X /var/task/function.zip .

tekumara

# ruff: noqa: SLF001 ARG001

from typing import Callable
from unittest.mock import MagicMock

import aiobotocore.awsrequest
import aiobotocore.endpoint
import aiohttp
import aiohttp.client_reqrep
import aiohttp.typedefs

tekumara

import base64
import math
from typing import Dict, cast

import cryptography.hazmat.backends.openssl.rsa as rsa
from cryptography.hazmat.primitives.serialization import load_der_private_key, load_der_public_key
from jose import jwt

# stolen from https://github.com/mpdavis/python-jose/blob/fccbcf4/tests/test_jws.py#L143
rsa_private_key = """-----BEGIN RSA PRIVATE KEY-----

tekumara

#!/usr/bin/env bash

set -uoe pipefail

function apt-get() {
    # wrap apt-get to handle lock contention due to unattended upgrades etc.
    # retry on lock contention up to three times. on other errors, exit without retry
    local i=0

    while true;

tekumara

from urllib.parse import urlparse

def parse_s3_url(url: str) -> tuple[str, str]:
    pr = urlparse(url)
    bucket = pr.netloc
    key = pr.path.lstrip("/")
    return bucket, key