Skip to content

Instantly share code, notes, and snippets.

import traceback
import sys
import inspect
import logging
import asyncio
# Настройка логирования
logging.basicConfig(level=logging.ERROR)
def custom_exception_handler(function):
@XanderMoroz
XanderMoroz / python_decorator_mesure_perfomance.py
Last active January 6, 2025 16:57
Асинхронный декоратор для измерения производительности функции
import tracemalloc
from time import perf_counter
from typing import Callable, TypeVar, Awaitable
from functools import wraps
F = TypeVar('F', bound=Callable[..., Awaitable[None]])
def measure_performance(func: F) -> F:
"""
@XanderMoroz
XanderMoroz / python_decorator_rate_limits.py
Created January 6, 2025 16:26
Асинхронный декоратор для ограничения количества вызовов функции за определенный период времени.
import time
import asyncio
from typing import Callable, TypeVar, Awaitable
F = TypeVar('F', bound=Callable[..., Awaitable[None]])
def rate_limits(max_calls: int, period: int) -> Callable[[F], F]:
"""
Асинхронный декоратор для ограничения количества вызовов функции за определенный период времени.
Attrs:
@XanderMoroz
XanderMoroz / python_decorator_retry.py
Created January 6, 2025 16:10
Декоратор запускающий функцию, и при возникновении ошибки повторяет несколько раз с задержкой
import asyncio
# Определение декоратора
def retry(max_retries: int, wait_time:int):
def decorator(func: Callable):
"""
Асинхронный декоратор осуществляющий несколько попыток выполнить функцию с задержкой
Attrs:
- max_retries: (int) — Количество попыток
- wait_time: (int) — Количество попыток
@XanderMoroz
XanderMoroz / python_useful_exception.py
Created January 6, 2025 15:16
Класс осуществляющий обработку исключений
class MyCustomException(Exception):
"""
Пример собственного исключения c информацией об ошибке
"""
def __init__(self, message, error_code, context_info=None, *args):
super().__init__(message, *args)
self.message = message
self.error_code = error_code
self.context_info = context_info
@XanderMoroz
XanderMoroz / python_redis_class.py
Last active August 17, 2025 03:18
Класс для взаимодействия с Redis на основе aio-redis. Может сохранять и отдавать по ключу атомарные данные и объекты.
"""Данный модуль отвечает за взаимодействие с Redis"""
import redis
import asyncio
import logging
from typing import Type, Union
from pydantic import BaseModel
from src.core.config import settings
logger = logging.getLogger(__name__)
@XanderMoroz
XanderMoroz / python_requester_class.py
Last active January 6, 2025 16:08
Класс осуществляющий подготовку и отправку запросов по HTTP на основе httpx. Может отправлять запросы в синхронном и асинхронном режимах
import httpx
import logging
from typing import Any
# Извлекаем объект для отслеживания логов
logger = logging.getLogger(__name__)
class Requester:
@XanderMoroz
XanderMoroz / python_rabbit_mq_class.py
Last active January 6, 2025 16:09
Класс для взаимодействия с RabbitMQ на основе aio-pika. Может создавать обменник (exchange) и очередь (queue), а потом связывать их. Может публиковать в очередь и извлекать из нее.
import logging
from typing import Type, Callable
from pydantic import BaseModel
import aio_pika
from config import RABBITMQ_USERNAME, RABBITMQ_PASSWORD, RABBITMQ_HOST, RABBITMQ_PORT
logger = logging.getLogger(__name__)
@XanderMoroz
XanderMoroz / python_useful_testing.py
Last active May 23, 2024 13:23
Инфраструктура для быстрого и удобного тестирования функциональности свежего кода. Тестирует синхронный и асинхронный код.
import asyncio
# Для тестирования
if __name__ == '__main__':
# Запускаем функцию синхронно
sync_result = some_sync_func(arg1=1, arg2=2)
# Выводим результат