nkhitrov

# TODO: fix high memory usage for rich models


class Base:
    @classmethod
    def get_real_column_name(cls, attr_name: str) -> str:
        return getattr(inspect(cls).c, attr_name).name

    def __str__(self) -> str:
        return self.__repr__()

nkhitrov

from loguru import logger


def console_formatter(record) -> str:
    # WARNING !!!
    # Функция должна возвращать строку, которая содержит только шаблоны для форматирования.
    # Если в строку прокидывать значения из record (или еще откуда-либо),
    # то loguru может принять их за f-строки и попытается обработать, что приведет к ошибке.
    # Например, если нужно достать какое-то значение из поля extra, вместо того чтобы прокидывать его в строку формата,
    # нужно прокидывать подстроку вида {extra[тут_ключ]}

nkhitrov

How to manually setup flake8 as PyCharm external tool

File / Settings / Tools / External Tools / Add
Name: Flake8
Program: $PyInterpreterDirectory$/python
Parameters: -m flake8 --max-complexity 10 --ignore E501 $FilePath$
Working directory: $ProjectFileDir$

Output Filters / Add
Name: Filter 1

nkhitrov

import contextvars
import time

from concurrent.futures import ThreadPoolExecutor, as_completed

current_user = contextvars.ContextVar("ID of current user")
shared_user = contextvars.ContextVar("ID of shared user")

def say_hello():
    time.sleep(1)

nkhitrov

import os
import logging.config
import structlog
from .app import app


timestamper = structlog.processors.TimeStamper(fmt="iso")
pre_chain = [
    # Add the log level and a timestamp to the event_dict if the log entry is not from structlog.
    structlog.stdlib.add_log_level,

nkhitrov

"""
Structlog example configuration with FastAPI.
Features:
- async bound logger
- contextvars to log request-id and other meta data
- custom format for default logging loggers and structlog loggers
"""

import asyncio
import logging

nkhitrov

import asyncio


async def run_task():
    print('run task!')
    await asyncio.sleep(3)
    print('long task finished!')


async def run_task_error():

nkhitrov

import asyncio
import functools
from typing import Any, Callable, Optional

from celery import shared_task
from rodi import Container, GetServiceContext

from redis.asyncio.cluster import RedisCluster
from sqlalchemy.ext.asyncio import AsyncSession

nkhitrov

from typing import Any, Generic, Type, TypeAlias, TypeVar

from fastapi import status as http_status_code
from pydantic import Field
from pydantic.generics import GenericModel


_Model = TypeVar("_Model")

GenericResponse: TypeAlias = GenericModel

nkhitrov

pip install virtualenv virtualenvwrapper

sudo echo '# where to store our virtual envs
export WORKON_HOME=$HOME/.pythonvenvs

# where is the virtualenvwrapper.sh
source $(which virtualenvwrapper.sh)
' >> ~/.bashrc

source ~/.bashrc