Cannot perform operation: another operation is in progress #728
Comments
|
@erhosen I faced the same problem, solution was to use a separate db connection: @pytest.fixture
async def special_model():
async with db.acquire(): # use seperate db connection
special_model = await SpecialModel.create(
internal_name='internal_name',
comment='comment',
is_active=True,
title="Title",
text="Text"
)
yield special_modelThis isn't an issue with |
|
Thanks! This deserves a page in the docs. |
|
I have the same exception Any idea what I did wrong? I would appreciate any help. Thanks in advance. requirements.txt test_code.py import os
from typing import List
import pytest
from gino import Gino
from fastapi import APIRouter
from pydantic import BaseModel
from fastapi import FastAPI
from starlette.testclient import TestClient
router = APIRouter()
db = Gino()
async def init_db():
await db.set_bind(os.environ['DATABASE_URL'])
await db.gino.create_all()
class UserModel(db.Model):
__tablename__ = 'user'
id = db.Column(db.Integer(), primary_key=True)
name = db.Column(db.Unicode())
email = db.Column(db.Unicode(), unique=True, index=True)
password_hash = db.Column(db.Unicode())
class UserSchema(BaseModel):
id: int = 0
name: str
email: str
password: str
class UserListSchema(BaseModel):
objects: List[UserSchema]
@router.get("/users/", response_model=UserListSchema)
async def get_users():
async with db.acquire():
users = await UserModel.query.limit(200).gino.all()
return UserListSchema.parse_obj({
'objects': [x.to_dict() for x in users]
})
def get_app():
print('INIT APP')
app = FastAPI(title="GINO FastAPI Demo")
app.include_router(router, prefix='/API/v1')
@app.on_event("startup")
async def startup_event():
print('Initialising DB')
await init_db()
print('DB was initialised')
return app
@pytest.fixture
def client():
with TestClient(get_app()) as client:
yield client
@pytest.fixture
@pytest.mark.anyio
async def user(client):
print('[-------->')
# await init_db()
# async with db.acquire():
user = await UserModel.create(email='[email protected]')
# async with db.acquire():
users = await UserModel.query.limit(200).gino.all()
print('.....=', user)
print('....._', users)
yield user
def test_users(user, client):
response = client.get(
"/API/v1/users",
headers={},
)
print('=====', user, response.text)
assert response.status_code == 200
assert response.json() == {} |
Sign up for free
to join this conversation on GitHub.
Already have an account?
Sign in to comment
Description
My task is create some database record (via GINO Model.create()) before running test (typical pytest.fixture usage), and drop everything after.
What I Did
conftest.py
test.py
What I got:
The text was updated successfully, but these errors were encountered: