victorusachev

def setup_audit(cls, level):
    # some code
    print(cls, level)


class AuditMeta(type):
    def __new__(mcs, name, bases, attrs, **kwargs):
        cls = super().__new__(mcs, name, bases, attrs)
        level = kwargs.get('level', 0)
        setup_audit(cls, level)

victorusachev

from typing import Any, Callable, Dict, Optional, Type, TypeVar

from environs import Env

__all__ = ('EnvVar',)

T = TypeVar('T')


class EnvVar:

victorusachev

from datetime import datetime
from sqlalchemy import Column, Integer, DateTime, ForeignKey
from sqlalchemy.orm import relationship
from sqlalchemy.ext.declarative import declared_attr
from flask_security import current_user


class AuditMixin(object):
    created_at = Column(DateTime, default=datetime.now)
    updated_at = Column(DateTime, default=datetime.now, onupdate=datetime.now)

victorusachev

import functools
import threading
from time import sleep

lock = threading.Lock()


class SingletonMeta(type):
    _instance = None
    _args = None

victorusachev

class SingletonMeta(type):
    _instance = None
    _args = None
    _kwargs = None

    def __call__(cls, *args, **kwargs):
        if not cls._instance:
            cls._args = args
            cls._kwargs = kwargs
            cls._instance = super(SingletonMeta, cls).__call__(*args, **kwargs)

victorusachev

SELECT
       mcl.changes #>> '{name, old_value, repr}'
FROM history_modelchangelog as mcl
WHERE (
    mcl.action = 'updated'
    AND (mcl.changes #>> '{name, new_value, repr}') = %(new_name)s
        AND (mcl.changes #>> '{name, old_value, repr}') IS NOT NULL
        AND mcl.content_type_id = %(content_type_id)s
        AND mcl.object_id = %(object_id)s
    )

victorusachev

from itertools import groupby
from operator import itemgetter


def get_subclasses(cls: type):
    for subclass in cls.__subclasses__():
        yield subclass
        yield from get_subclasses(subclass)

victorusachev

"""
My solution

See https://t.me/aiohttp_ru/3955
"""
import asyncio
import json

import aiohttp
from asgiref.sync import sync_to_async

victorusachev

from urllib.parse import (
    parse_qsl,
    urlencode,
    urlparse,
    urlunparse,
)


def inject_query_parameters(url: str, params: dict) -> str:
    parts = urlparse(url)

victorusachev

from collections import OrderedDict

empty = object()


class Item:
    """
    Descriptor
    Usage:
     >>>    class SomethingType: