matanper

@app.middleware("http")
async def add_sql_tap(request: Request, call_next):
    profiler = sqltap.start()
    response = await call_next(request)
    statistics = profiler.collect()
    sqltap.report(statistics, "report.txt", report_format="text")
    return response

matanper

import signal
import asyncio

from prometheus_client import start_http_server as start_prometheus_server


EXIT_EVENT = asyncio.Event()


async def infinite_run():

matanper

import valkey

class PrefixedValkey(valkey.Valkey):  # Use `ValkeyCluster` if needed
    def __init__(self, *args, key_prefix: str, **kwargs):
        super().__init__(*args, **kwargs)
        self.key_prefix = key_prefix

    def _prefix_key(self, key):
        """Prefix Redis keys with the worker-specific namespace."""
        if isinstance(key, bytes):  # Handle binary keys

matanper

select pid, 
       usename, 
       pg_blocking_pids(pid) as blocked_by, 
       query as blocked_query
from pg_stat_activity
where cardinality(pg_blocking_pids(pid)) > 0;

matanper

import asyncio
import logging
from asyncio import AbstractEventLoop
from concurrent.futures import Future
from threading import Thread
from typing import Coroutine

from app.common.logger import logger

matanper

package org.monocle;

import java.io.IOException;

import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;

matanper

import boto3
import os
import time
from subprocess import Popen, PIPE
import pyarrow.parquet as pq
import pickle
import hashlib


TSDB_CPUS = 8

matanper

def to_asyncio_in_thread(func):
    async def wrapper(*args, **kwargs):
        return await asyncio.get_event_loop().run_in_executor(None, lambda: func(*args, **kwargs))

    return wrapper


 # Usage: 
  
@to_asyncio_in_thread

matanper

from cachetools import TTLCache
from cachetools.keys import hashkey

from parameterized_lock import parameterized_lock


def async_threadsafe_ttl_cache(func=None, ttl=60):
    cache = TTLCache(maxsize=100, ttl=ttl)

    def decorator(decorated_func):

matanper

import cProfile
import io
import pstats

pr = cProfile.Profile()
pr.enable()
### Code to profile here
pr.disable()
s = io.StringIO()
ps = pstats.Stats(pr, stream=s).sort_stats('time')