Created
November 18, 2021 17:48
-
-
Save kgantsov/27b1b88e9da5298c9143e0090860a315 to your computer and use it in GitHub Desktop.
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
fastapi==0.70.0 | |
gino==1.0.1 | |
pytest==6.2.5 | |
pytest-asyncio==0.16.0 | |
requests==2.26.0 |
This file contains hidden or bidirectional Unicode text that may be interpreted or compiled differently than what appears below. To review, open the file in an editor that reveals hidden Unicode characters.
Learn more about bidirectional Unicode characters
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