Skip to content

Instantly share code, notes, and snippets.

@kgantsov
Created November 18, 2021 17:48
Show Gist options
  • Save kgantsov/27b1b88e9da5298c9143e0090860a315 to your computer and use it in GitHub Desktop.
Save kgantsov/27b1b88e9da5298c9143e0090860a315 to your computer and use it in GitHub Desktop.
fastapi==0.70.0
gino==1.0.1
pytest==6.2.5
pytest-asyncio==0.16.0
requests==2.26.0
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='test@gmail.com')
# 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