Connected MySQL using Peeweee

Author: s3m3dov

Dockerfile

@@ -10,7 +10,7 @@ WORKDIR /code
 
 # Install dependencies
 COPY ./requirements.txt /code/requirements.txt
-RUN pip install --no-cache-dir --upgrade -r /code/requirements.txt
+RUN pip install --upgrade -r /code/requirements.txt
 
 # Copy project
 COPY . /code/

docker-compose.yml

@@ -8,6 +8,7 @@ services:
       - .:/code
     ports:
       - "8000:8000"
+
     depends_on:
       - db
 
@@ -16,20 +17,26 @@ services:
     restart: always
     tty: true
     environment:
-      MYSQL_DATABASE: mysql_db
-      MYSQL_USER: mysql
-      MYSQL_PASSWORD: mysql
-      MYSQL_ROOT_PASSWORD: mysql
+      - MYSQL_DATABASE=mysql_db
+      - MYSQL_USER=mysql
+      - MYSQL_PASSWORD=mysql
+      - MYSQL_ROOT_PASSWORD=mysql
+      - MYSQL_HOST=db
+      - MYSQL_PORT=3307
     ports:
       - "3307:3306"
+
     volumes:
       - mysql_data:/var/lib/mysql/
 
-  adminer:
-    image: adminer:latest
+  phpmyadmin:
+    image: phpmyadmin:latest
     restart: always
     ports:
-      - "8080:8080"
+      - "8080:80"
+    environment:
+      - PMA_ARBITRARY=1
 
 volumes:
   mysql_data:
+

main.py

@@ -1,9 +1,36 @@
+import os, time
+from typing import List
+
 import uvicorn
-from fastapi import FastAPI
+from fastapi import Depends, FastAPI, HTTPException
+
+from sql_app import crud, schemas, database, models
+from sql_app.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()
+
+
+# Endpoints
 @app.get("/")
 async def root():
     return {"message": "Hello World"}
@@ -14,5 +41,56 @@ async def say_hello(name: str):
     return {"message": f"Hello {name}"}
 
 
+@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
+
+
+# Run Fast API app
 if __name__ == "__main__":
     uvicorn.run(app, host="0.0.0.0", port=8000)

requirements.txt

@@ -1,2 +1,8 @@
 fastapi~=0.73.0
-uvicorn~=0.17.1
+uvicorn~=0.17.3
+peewee~=3.14.8
+pydantic~=1.9.0
+contextvars~=2.4.0
+pymysql~=1.0.2
+cryptography~=36.0.1
+environs~=9.5.0

sql_app/__init__.py

@@ -0,0 +1,30 @@
+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

sql_app/crud.py

@@ -0,0 +1,37 @@
+from environs import Env
+from contextvars import ContextVar
+
+import peewee
+
+# Environment
+env = Env()
+env.read_env()
+
+DATABASE_NAME = env("MYSQL_DATABASE", default="mysql_db")
+DATABASE_USER = env("MYSQL_USER", default="mysql")
+DATABASE_PASS = env("MYSQL_PASSWORD", default="mysql")
+DATABASE_PORT = env("MYSQL_HOST", default="db")
+DATABASE_PORT = env.int("MYSQL_PORT", default=3306)
+
+
+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.MySQLDatabase(DATABASE_NAME, user=DATABASE_USER, password=DATABASE_PASS, host="db", port=DATABASE_PORT)
+
+db._state = PeeweeConnectionState()
+
+# Peewee was not designed for async frameworks, or with them in mind.

sql_app/database.py

@@ -0,0 +1,20 @@
+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

sql_app/models.py

@@ -0,0 +1,49 @@
+from typing import Any, List, Optional
+
+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: Optional[str] = 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