nkhitrov

"""
WARNING: dont use loguru, use structlog
https://gist.github.com/nkhitrov/38adbb314f0d35371eba4ffb8f27078f


Configure handlers and formats for application loggers.
"""
import logging
import sys
from pprint import pformat

nkhitrov

#!/usr/bin/python
# A Wake on LAN program that allows you to send magic packets over the Internet
import socket, struct

class Waker():
    def makeMagicPacket(self, macAddress):
        # Take the entered MAC address and format it to be sent via socket
        splitMac = str.split(macAddress,':')
    
        # Pack together the sections of the MAC address as binary hex

nkhitrov

from typing import Optional
import base64
from passlib.context import CryptContext
from datetime import datetime, timedelta

import jwt
from jwt import PyJWTError

from pydantic import BaseModel

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

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

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

import asyncio


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


async def run_task_error():

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 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

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)