Вот мой код в файле main.py:
Код: Выделить всё
from fastapi import FastAPI
from fastapi.testclient import TestClient
from contextlib import asynccontextmanager
from .internals.settings import get_config
from .internals.tags import tags_metadata
from .internals.postgres import init_postgres
from .services.reservation import init_sqllite, reload_seat_to_unlock
import aiomysql
@asynccontextmanager
async def lifespan(app: FastAPI):
config = get_config()
app.state.pool = await aiomysql.create_pool(
host=config.APP_DB_HOST,
port=config.APP_DB_PORT,
user=config.APP_DB_USER,
password=config.APP_DB_PASSWORD,
db=config.APP_DB_DATABASE,
)
init_sqllite(app)
reload_seat_to_unlock(app)
init_postgres(app)
yield
app = FastAPI(
lifespan=lifespan,
title="Actifastage",
summary="Actistages JsonAPI",
version="1.3.2",
contact={
"name": "educacode",
"url": "https://www.educacode.com",
"email": "contact@educacode.com",
},
openapi_tags=tags_metadata,
)
client = TestClient(app)
Код: Выделить всё
from app.main import client
async def test_login(user={"username": "unit-test", "password": "unittest"}):
response = client.post("/token", data=user)
assert response.status_code == 200
AttributeError: у объекта «Состояние» нет атрибута «пул»
p>
Вот пример запроса к базе данных, который я делаю:
Код: Выделить всё
async def get_user(username: str) -> Optional[UserInDB]:
async with app.state.pool.acquire() as conn:
async with conn.cursor(aiomysql.DictCursor) as cur:
await cur.execute(
"SELECT id, username, password, is_active FROM sf_guard_user WHERE username=%s AND is_active=TRUE",
username,
)
res = await cur.fetchone()
await conn.commit()
res = UserInDB.read(res) if res else None
return res
Спасибо
Я пытаюсь повторно объявить продолжительность жизни, чтобы инициировать новое app.state .pool но у меня уже такая же проблема.
Подробнее здесь: https://stackoverflow.com/questions/792 ... th-fastapi
Мобильная версия