Skip to content

Instantly share code, notes, and snippets.

from logging.config import dictConfig
LOG_CONFIG = {
"version": 1,
"disable_existing_loggers": False,
"filters": {
"require_debug_true": {
"()": "core.config.logging.filters.IsDebugMode",
},
"require_debug_false": {
@Niccolum
Niccolum / conftest.py
Created June 14, 2022 06:22
How to prepare conftest for testing fastapi with sqlmodel
import asyncio
import logging
import typing as t
from contextlib import contextmanager
import fastapi
import pytest
from httpx import AsyncClient
from mixer.backend.sqlalchemy import Mixer
from pytest_mock import MockerFixture

Flake8

pep8 warning

E***

  • E1** Indentation
  • E2** Whitespace
  • E3** Blank line
  • E4** Import
from concurrent.futures import ProcessPoolExecutor
import multiprocessing as mp
import time
import random
PROCESSES = 3
def main_proc(queues):
kosmos_cron | start at 2020-06-19 07:12:01.918786
kosmos_cron | http://kosmos-app:8080
kosmos_cron | Traceback (most recent call last):
kosmos_cron | File "/usr/local/lib/python3.7/site-packages/urllib3/connection.py", line 160, in _new_conn
kosmos_cron | (self._dns_host, self.port), self.timeout, **extra_kw
kosmos_cron | File "/usr/local/lib/python3.7/site-packages/urllib3/util/connection.py", line 61, in create_connection
kosmos_cron | for res in socket.getaddrinfo(host, port, family, socket.SOCK_STREAM):
kosmos_cron | File "/usr/local/lib/python3.7/socket.py", line 752, in getaddrinfo
kosmos_cron | for res in _socket.getaddrinfo(host, port, family, type, proto, flags):
kosmos_cron | socket.gaierror: [Errno -3] Temporary failure in name resolution
def request_schema(schema, locations=None, put_into=None, example=None, add_to_refs=False, **kwargs):
if callable(schema):
schema = schema()
# location kwarg added for compatibility with old versions
locations = locations or []
if not locations:
locations = kwargs.pop("location", None)
if locations:
locations = [locations]
else:
with user_answers as (
select
qa.uuid as questions_answers_id,
CASE
when st.grade::int = 1 then TRUE
else FALSE
END as grade,
CASE
when st.state = 'checked' then TRUE
else FALSE
# start_func.py
async def start_func(testcls, first, **second):
second['new_arg'] = 'wow'
await testcls.send(first, **second)
if __name__ == '__main__':
from end_func import TestCls
import asyncio