Add application

examples/miniapps/fastapi-sqlalchemy/config.yml

@@ -0,0 +1,2 @@
+db:
+  url: "sqlite:///./sql_app.db"

examples/miniapps/fastapi-sqlalchemy/requirements.txt

@@ -0,0 +1,5 @@
+dependency-injector
+fastapi
+uvicorn
+pyyaml
+sqlalchemy

examples/miniapps/fastapi-sqlalchemy/webapp/__init__.py

@@ -0,0 +1 @@
+"""Top-level package."""

examples/miniapps/fastapi-sqlalchemy/webapp/application.py

@@ -0,0 +1,23 @@
+"""Application module."""
+
+from fastapi import FastAPI
+
+from .containers import Container
+from . import endpoints
+
+
+def create_app() -> FastAPI:
+    container = Container()
+    container.config.from_yaml('config.yml')
+    container.wire(modules=[endpoints])
+
+    db = container.db()
+    db.create_database()
+
+    app = FastAPI()
+    app.container = container
+    app.include_router(endpoints.router)
+    return app
+
+
+app = create_app()

examples/miniapps/fastapi-sqlalchemy/webapp/containers.py

@@ -0,0 +1,24 @@
+"""Containers module."""
+
+from dependency_injector import containers, providers
+
+from .database import Database
+from .repositories import UserRepository
+from .services import UserService
+
+
+class Container(containers.DeclarativeContainer):
+
+    config = providers.Configuration()
+
+    db = providers.Singleton(Database, db_url=config.db.url)
+
+    users_repository = providers.Factory(
+        UserRepository,
+        session_factory=db.provided.session,
+    )
+
+    user_service = providers.Factory(
+        UserService,
+        users_repository=users_repository,
+    )

examples/miniapps/fastapi-sqlalchemy/webapp/database.py

@@ -0,0 +1,41 @@
+"""Database module."""
+
+from contextlib import contextmanager, AbstractContextManager
+from typing import Callable
+import logging
+
+from sqlalchemy import create_engine, orm
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import Session
+
+logger = logging.getLogger(__name__)
+
+Base = declarative_base()
+
+
+class Database:
+
+    def __init__(self, db_url: str) -> None:
+        self._engine = create_engine(db_url, echo=True)
+        self._session_factory = orm.scoped_session(
+            orm.sessionmaker(
+                autocommit=False,
+                autoflush=False,
+                bind=self._engine,
+            ),
+        )
+
+    def create_database(self) -> None:
+        Base.metadata.create_all(self._engine)
+
+    @contextmanager
+    def session(self) -> Callable[..., AbstractContextManager[Session]]:
+        session: Session = self._session_factory()
+        try:
+            yield session
+        except Exception:
+            logger.exception('Session rollback because of exception')
+            session.rollback()
+            raise
+        finally:
+            session.close()

examples/miniapps/fastapi-sqlalchemy/webapp/endpoints.py

@@ -0,0 +1,57 @@
+"""Endpoints module."""
+
+from fastapi import APIRouter, Depends, Response, status
+from dependency_injector.wiring import inject, Provide
+
+from .containers import Container
+from .services import UserService
+from .repositories import NotFoundError
+
+router = APIRouter()
+
+
+@router.get('/users')
+@inject
+def get_list(
+        user_service: UserService = Depends(Provide[Container.user_service]),
+):
+    return user_service.get_users()
+
+
+@router.get('/users/{user_id}')
+@inject
+def get_by_id(
+        user_id: int,
+        response: Response,
+        user_service: UserService = Depends(Provide[Container.user_service]),
+):
+    try:
+        return user_service.get_user_by_id(user_id)
+    except NotFoundError:
+        response.status_code = status.HTTP_404_NOT_FOUND
+
+
+@router.post('/users', status_code=status.HTTP_201_CREATED)
+@inject
+def add(
+        user_service: UserService = Depends(Provide[Container.user_service]),
+):
+    return user_service.create_user()
+
+
+@router.delete('/users/{user_id}', status_code=status.HTTP_204_NO_CONTENT)
+@inject
+def remove(
+        user_id: int,
+        response: Response,
+        user_service: UserService = Depends(Provide[Container.user_service]),
+):
+    try:
+        user_service.delete_user_by_id(user_id)
+    except NotFoundError:
+        response.status_code = status.HTTP_404_NOT_FOUND
+
+
+@router.get('/status')
+def get_status():
+    return {'status': 'OK'}

examples/miniapps/fastapi-sqlalchemy/webapp/models.py

@@ -0,0 +1,21 @@
+"""Models module."""
+
+from sqlalchemy import Column, String, Boolean, Integer
+
+from .database import Base
+
+
+class User(Base):
+
+    __tablename__ = 'users'
+
+    id = Column(Integer, primary_key=True)
+    email = Column(String, unique=True)
+    hashed_password = Column(String)
+    is_active = Column(Boolean, default=True)
+
+    def __repr__(self):
+        return f'<User(id="{self.id}", ' \
+               f'email="{self.email}", ' \
+               f'hashed_password="{self.hashed_password}", ' \
+               f'is_active="{self.is_active}")>'

examples/miniapps/fastapi-sqlalchemy/webapp/repositories.py

@@ -0,0 +1,54 @@
+"""Repositories module."""
+
+from contextlib import AbstractContextManager
+from typing import Callable, Iterator
+
+from sqlalchemy.orm import Session
+
+from .models import User
+
+
+class UserRepository:
+
+    def __init__(self, session_factory: Callable[..., AbstractContextManager[Session]]) -> None:
+        self.session_factory = session_factory
+
+    def get_all(self) -> Iterator[User]:
+        with self.session_factory() as session:
+            return session.query(User).all()
+
+    def get_by_id(self, user_id: int) -> User:
+        with self.session_factory() as session:
+            user = session.query(User).filter(User.id == user_id).first()
+            if not user:
+                raise UserNotFoundError(user_id)
+            return user
+
+    def add(self, email: str, password: str, is_active: bool = True) -> User:
+        with self.session_factory() as session:
+            user = User(email=email, hashed_password=password, is_active=is_active)
+            session.add(user)
+            session.commit()
+            session.refresh(user)
+            return user
+
+    def delete_by_id(self, user_id: int) -> None:
+        with self.session_factory() as session:
+            entity: User = session.query(User).filter(User.id == user_id).first()
+            if not entity:
+                raise UserNotFoundError(user_id)
+            session.delete(entity)
+            session.commit()
+
+
+class NotFoundError(Exception):
+
+    entity_name: str
+
+    def __init__(self, entity_id):
+        super().__init__(f'{self.entity_name} not found, id: {entity_id}')
+
+
+class UserNotFoundError(NotFoundError):
+
+    entity_name: str = 'User'

examples/miniapps/fastapi-sqlalchemy/webapp/services.py

@@ -0,0 +1,26 @@
+"""Services module."""
+
+from uuid import uuid4
+from typing import Iterator
+
+from .repositories import UserRepository
+from .models import User
+
+
+class UserService:
+
+    def __init__(self, users_repository: UserRepository) -> None:
+        self._repository: UserRepository = users_repository
+
+    def get_users(self) -> Iterator[User]:
+        return self._repository.get_all()
+
+    def get_user_by_id(self, user_id: int) -> User:
+        return self._repository.get_by_id(user_id)
+
+    def create_user(self) -> User:
+        uid = uuid4()
+        return self._repository.add(email=f'{uid}@email.com', password='pwd')
+
+    def delete_user_by_id(self, user_id: int) -> None:
+        return self._repository.delete_by_id(user_id)