MySQL database

.env

@@ -1,5 +1,5 @@
-MYSQL_HOST=${REPO_NAME_SNAKE}_db
+DB_HOST=${REPO_NAME_SNAKE}_db
-MYSQL_PORT=3306
+DB_PORT=3306
-MYSQL_USER=${REPO_NAME_SNAKE}
+DB_USER=${REPO_NAME_SNAKE}
-MYSQL_PASSWORD=
+DB_PASSWORD=
-MYSQL_DATABASE=${REPO_NAME_SNAKE}
+DB_DATABASE=${REPO_NAME_SNAKE}

.env_db

@@ -0,0 +1,5 @@
+MYSQL_HOST=${REPO_NAME_SNAKE}_db
+MYSQL_PORT=3306
+MYSQL_USER=${REPO_NAME_SNAKE}
+MYSQL_PASSWORD=
+MYSQL_DATABASE=${REPO_NAME_SNAKE}

.env_debug

@@ -1,5 +1,5 @@
-MYSQL_HOST=localhost
+DB_HOST=localhost
-MYSQL_PORT=3306
+DB_PORT=3306
-MYSQL_USER=darkcat09
+DB_USER=darkcat09
-MYSQL_PASSWORD=
+DB_PASSWORD=
-MYSQL_DATABASE=apptest
+DB_DATABASE=apptest

app/sql/crud.py

@@ -0,0 +1,38 @@
+from typing import Optional, List
+from sqlalchemy.orm import Session
+
+from . import models
+from . import schemas
+
+
+def get_user(
+        db: Session,
+        user_id: int) -> Optional[models.User]:
+
+    return db \
+        .query(models.User) \
+        .filter(models.User.id == user_id) \
+        .first()
+
+
+def get_users(
+        db: Session,
+        skip: int = 0,
+        limit: int = 100) -> List[Optional[models.User]]:
+
+    return db \
+        .query(models.User) \
+        .offset(skip) \
+        .limit(limit) \
+        .all()
+
+
+def create_user(
+        db: Session,
+        user: schemas.User) -> models.User:
+
+    user_model = models.User(**user.dict())
+    db.add(user_model)
+    db.commit()
+    db.refresh(user_model)
+    return user_model

app/sql/db.py

@@ -0,0 +1,46 @@
+from typing import Generator
+from pydantic import BaseSettings
+
+from sqlalchemy import create_engine
+from sqlalchemy.orm import Session, sessionmaker
+from sqlalchemy.ext.declarative import declarative_base
+
+
+class SqlSettings(BaseSettings):
+    db_host: str = '${REPO_NAME_SNAKE}_db'
+    db_port: int = 3306
+    db_user: str = '${REPO_NAME_SNAKE}'
+    db_password: str = ''
+    db_database: str = '${REPO_NAME_SNAKE}'
+
+sql_settings = SqlSettings()
+
+
+db_url = (
+    'mysql://{db_user}:{db_password}@'
+    '{db_host}:{db_port}/${db_database}'
+).format(**sql_settings.dict())
+
+engine = create_engine(db_url)
+SessionLocal = sessionmaker(
+    autocommit=False,
+    autoflush=False,
+    bind=engine,
+)
+Base = declarative_base()
+
+
+def get_db() -> Generator[Session, None, None]:
+    """FastAPI dependency
+    returning database Session object.
+    Code is copied from the official docs
+
+    Yields:
+        SQLAlchemy Session object
+    """
+
+    db = SessionLocal()
+    try:
+        yield db
+    finally:
+        db.close()

app/sql/models.py

@@ -0,0 +1,12 @@
+from sqlalchemy import Column, String, Integer
+
+from .db import Base
+
+
+class User(Base):
+    __tablename__ = 'users'
+
+    id = Column(Integer, primary_key=True)
+    email = Column(String)
+    name = Column(String)
+    age = Column(Integer)

app/sql/schemas.py

@@ -0,0 +1,11 @@
+from pydantic import BaseModel
+
+
+class User(BaseModel):
+    id: int
+    email: str
+    name: str
+    age: int
+
+    class Config:
+        orm_mode = True