+++ /dev/null
-from . import models, schemas
-
-
-def get_user(user_id: int):
- return models.User.filter(models.User.id == user_id).first()
-
-
-def get_user_by_email(email: str):
- return models.User.filter(models.User.email == email).first()
-
-
-def get_users(skip: int = 0, limit: int = 100):
- return list(models.User.select().offset(skip).limit(limit))
-
-
-def create_user(user: schemas.UserCreate):
- fake_hashed_password = user.password + "notreallyhashed"
- db_user = models.User(email=user.email, hashed_password=fake_hashed_password)
- db_user.save()
- return db_user
-
-
-def get_items(skip: int = 0, limit: int = 100):
- return list(models.Item.select().offset(skip).limit(limit))
-
-
-def create_user_item(item: schemas.ItemCreate, user_id: int):
- db_item = models.Item(**item.dict(), owner_id=user_id)
- db_item.save()
- return db_item
+++ /dev/null
-from contextvars import ContextVar
-
-import peewee
-
-DATABASE_NAME = "test.db"
-db_state_default = {"closed": None, "conn": None, "ctx": None, "transactions": None}
-db_state = ContextVar("db_state", default=db_state_default.copy())
-
-
-class PeeweeConnectionState(peewee._ConnectionState):
- def __init__(self, **kwargs):
- super().__setattr__("_state", db_state)
- super().__init__(**kwargs)
-
- def __setattr__(self, name, value):
- self._state.get()[name] = value
-
- def __getattr__(self, name):
- return self._state.get()[name]
-
-
-db = peewee.SqliteDatabase(DATABASE_NAME, check_same_thread=False)
-
-db._state = PeeweeConnectionState()
+++ /dev/null
-import time
-from typing import List
-
-from fastapi import Depends, FastAPI, HTTPException
-
-from . import crud, database, models, schemas
-from .database import db_state_default
-
-database.db.connect()
-database.db.create_tables([models.User, models.Item])
-database.db.close()
-
-app = FastAPI()
-
-sleep_time = 10
-
-
-async def reset_db_state():
- database.db._state._state.set(db_state_default.copy())
- database.db._state.reset()
-
-
-def get_db(db_state=Depends(reset_db_state)):
- try:
- database.db.connect()
- yield
- finally:
- if not database.db.is_closed():
- database.db.close()
-
-
-@app.post("/users/", response_model=schemas.User, dependencies=[Depends(get_db)])
-def create_user(user: schemas.UserCreate):
- db_user = crud.get_user_by_email(email=user.email)
- if db_user:
- raise HTTPException(status_code=400, detail="Email already registered")
- return crud.create_user(user=user)
-
-
-@app.get("/users/", response_model=List[schemas.User], dependencies=[Depends(get_db)])
-def read_users(skip: int = 0, limit: int = 100):
- users = crud.get_users(skip=skip, limit=limit)
- return users
-
-
-@app.get(
- "/users/{user_id}", response_model=schemas.User, dependencies=[Depends(get_db)]
-)
-def read_user(user_id: int):
- db_user = crud.get_user(user_id=user_id)
- if db_user is None:
- raise HTTPException(status_code=404, detail="User not found")
- return db_user
-
-
-@app.post(
- "/users/{user_id}/items/",
- response_model=schemas.Item,
- dependencies=[Depends(get_db)],
-)
-def create_item_for_user(user_id: int, item: schemas.ItemCreate):
- return crud.create_user_item(item=item, user_id=user_id)
-
-
-@app.get("/items/", response_model=List[schemas.Item], dependencies=[Depends(get_db)])
-def read_items(skip: int = 0, limit: int = 100):
- items = crud.get_items(skip=skip, limit=limit)
- return items
-
-
-@app.get(
- "/slowusers/", response_model=List[schemas.User], dependencies=[Depends(get_db)]
-)
-def read_slow_users(skip: int = 0, limit: int = 100):
- global sleep_time
- sleep_time = max(0, sleep_time - 1)
- time.sleep(sleep_time) # Fake long processing request
- users = crud.get_users(skip=skip, limit=limit)
- return users
+++ /dev/null
-import peewee
-
-from .database import db
-
-
-class User(peewee.Model):
- email = peewee.CharField(unique=True, index=True)
- hashed_password = peewee.CharField()
- is_active = peewee.BooleanField(default=True)
-
- class Meta:
- database = db
-
-
-class Item(peewee.Model):
- title = peewee.CharField(index=True)
- description = peewee.CharField(index=True)
- owner = peewee.ForeignKeyField(User, backref="items")
-
- class Meta:
- database = db
+++ /dev/null
-from typing import Any, List, Union
-
-import peewee
-from pydantic import BaseModel
-from pydantic.utils import GetterDict
-
-
-class PeeweeGetterDict(GetterDict):
- def get(self, key: Any, default: Any = None):
- res = getattr(self._obj, key, default)
- if isinstance(res, peewee.ModelSelect):
- return list(res)
- return res
-
-
-class ItemBase(BaseModel):
- title: str
- description: Union[str, None] = None
-
-
-class ItemCreate(ItemBase):
- pass
-
-
-class Item(ItemBase):
- id: int
- owner_id: int
-
- class Config:
- orm_mode = True
- getter_dict = PeeweeGetterDict
-
-
-class UserBase(BaseModel):
- email: str
-
-
-class UserCreate(UserBase):
- password: str
-
-
-class User(UserBase):
- id: int
- is_active: bool
- items: List[Item] = []
-
- class Config:
- orm_mode = True
- getter_dict = PeeweeGetterDict