[ISSUE-33] Add clients with cache and reduce

Author: andy-takker

.env.dev

@@ -1 +1,6 @@
-APP_DB_DSN=
+APP_DB_DSN=
+APP_DB_POOL_SIZE=
+APP_DB_POOL_TIMEOUT=
+APP_DB_MAX_OVERFLOW=
+
+APP_REDIS_DSN=

.gitignore

@@ -159,3 +159,4 @@ cython_debug/
 #  option (not recommended) you can uncomment the following to ignore the entire idea folder.
 .idea/
 .vscode/
+.DS_Store

docker-compose.dev.yaml

@@ -1,4 +1,8 @@
 services:
+  redis:
+    image: redis:latest
+    ports:
+      - "6379:6379"
   db:
     image: postgres:16
     ports:

docker-compose.yaml

@@ -16,6 +16,11 @@ services:
     volumes:
       - postgres_data:/var/lib/postgresql/data
 
+  redis:
+    image: redis:latest
+    ports:
+      - "${REDIS_PORT}:6379"
+
   app:
     image: andytakker/example-web-service:latest
     command: granian --interface asgi --host 0.0.0.0 --port 8080 --loop uvloop library.main:app --access-log --log-level info

library/adapters/database/alembic.ini

@@ -2,7 +2,7 @@
 script_location = %(here)s/migrations
 file_template = %%(year)d_%%(month).2d_%%(day).2d_%%(rev)s_%%(slug)s
 prepend_sys_path = .
-version_path_separator = os
+path_separator = os
 
 [post_write_hooks]
 
@@ -39,4 +39,3 @@ formatter = generic
 [formatter_generic]
 format = %(levelname)-5.5s [%(name)s] %(message)s
 datefmt = %H:%M:%S
-

library/adapters/database/config.py

@@ -11,3 +11,6 @@ class DatabaseConfig:
     pool_timeout: int = field(
         default_factory=lambda: int(environ.get("APP_DB_POOL_TIMEOUT", 10))
     )
+    max_overflow: int = field(
+        default_factory=lambda: int(environ.get("APP_DB_MAX_OVERFLOW", 10))
+    )

library/adapters/database/di.py

@@ -17,16 +17,28 @@ def __init__(
         self,
         dsn: str,
         debug: bool,
+        pool_size: int,
+        pool_timeout: int,
+        max_overflow: int,
         scope: BaseScope | None = None,
         component: Component | None = None,
     ) -> None:
-        self.dsn = dsn
-        self.debug = debug
+        self._dsn = dsn
+        self._debug = debug
+        self._pool_size = pool_size
+        self._pool_timeout = pool_timeout
+        self._max_overflow = max_overflow
         super().__init__(scope=scope, component=component)
 
     @provide(scope=Scope.APP)
     async def engine(self) -> AsyncIterator[AsyncEngine]:
-        async with create_engine(dsn=self.dsn, debug=self.debug) as engine:
+        async with create_engine(
+            dsn=self._dsn,
+            debug=self._debug,
+            pool_size=self._pool_size,
+            pool_timeout=self._pool_timeout,
+            max_overflow=self._max_overflow,
+        ) as engine:
             yield engine
 
     @provide(scope=Scope.APP)
@@ -37,12 +49,12 @@ def session_factory(self, engine: AsyncEngine) -> async_sessionmaker[AsyncSessio
     def uow(
         self, session_factory: async_sessionmaker[AsyncSession]
     ) -> AnyOf[SqlalchemyUow, AbstractUow]:
-        return SqlalchemyUow(session=session_factory())
+        return SqlalchemyUow(session_factory=session_factory)
 
     @provide(scope=Scope.REQUEST)
     def book_storage(self, uow: SqlalchemyUow) -> IBookStorage:
-        return BookStorage(session=uow.session)
+        return BookStorage(uow=uow)
 
     @provide(scope=Scope.REQUEST)
     def user_storage(self, uow: SqlalchemyUow) -> IUserStorage:
-        return UserStorage(session=uow.session)
+        return UserStorage(uow=uow)

library/adapters/database/storages/book.py

@@ -7,6 +7,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from library.adapters.database.tables import BookTable
+from library.adapters.database.uow import SqlalchemyUow
 from library.application.exceptions import (
     EntityAlreadyExistsException,
     EntityNotFoundException,
@@ -19,19 +20,22 @@
     CreateBook,
     UpdateBook,
 )
-from library.domains.interfaces.storages.book import IBookStorage
 
 
-class BookStorage(IBookStorage):
-    def __init__(self, session: AsyncSession) -> None:
-        self.session = session
+class BookStorage:
+    def __init__(self, *, uow: SqlalchemyUow) -> None:
+        self._uow = uow
+
+    @property
+    def _session(self) -> AsyncSession:
+        return self._uow.session
 
     async def fetch_book_by_id(self, *, book_id: BookId) -> Book | None:
         query = select(BookTable).where(
             BookTable.id == book_id,
             BookTable.deleted_at.is_(None),
         )
-        book = (await self.session.scalars(query)).first()
+        book = (await self._session.scalars(query)).first()
 
         if book is None:
             return None
@@ -48,15 +52,15 @@ async def exists_book_by_id(self, *, book_id: BookId) -> bool:
         stmt = select(
             exists().where(BookTable.id == book_id, BookTable.deleted_at.is_(None))
         )
-        return bool((await self.session.execute(stmt)).scalar())
+        return bool((await self._session.execute(stmt)).scalar())
 
     async def count_books(self, *, params: BookPaginationParams) -> int:
         query = (
             select(func.count())
             .select_from(BookTable)
             .where(BookTable.deleted_at.is_(None))
         )
-        result = (await self.session.execute(query)).scalar()
+        result = (await self._session.execute(query)).scalar()
         return result or 0
 
     async def fetch_book_list(self, *, params: BookPaginationParams) -> Sequence[Book]:
@@ -74,7 +78,7 @@ async def fetch_book_list(self, *, params: BookPaginationParams) -> Sequence[Boo
             .offset(params.offset)
             .order_by(BookTable.id)
         )
-        result = (await self.session.execute(query)).mappings().all()
+        result = (await self._session.execute(query)).mappings().all()
         return [
             Book(
                 id=book["id"],
@@ -105,7 +109,7 @@ async def create_book(self, *, book: CreateBook) -> Book:
             )
         )
         try:
-            result = (await self.session.execute(stmt)).mappings().one()
+            result = (await self._session.execute(stmt)).mappings().one()
         except IntegrityError as e:
             self._raise_error(e)
         return Book(
@@ -123,7 +127,7 @@ async def delete_book_by_id(self, *, book_id: BookId) -> None:
             .where(BookTable.id == book_id)
             .values(deleted_at=datetime.now(tz=UTC))
         )
-        await self.session.execute(stmt)
+        await self._session.execute(stmt)
 
     async def update_book_by_id(self, *, update_book: UpdateBook) -> Book:
         stmt = (
@@ -140,7 +144,7 @@ async def update_book_by_id(self, *, update_book: UpdateBook) -> Book:
             )
         )
         try:
-            result = (await self.session.execute(stmt)).mappings().one()
+            result = (await self._session.execute(stmt)).mappings().one()
         except NoResultFound as e:
             raise EntityNotFoundException(entity=Book, entity_id=update_book.id) from e
         except IntegrityError as e:

library/adapters/database/storages/user.py

@@ -7,6 +7,7 @@
 from sqlalchemy.ext.asyncio import AsyncSession
 
 from library.adapters.database.tables import UserTable
+from library.adapters.database.uow import SqlalchemyUow
 from library.application.exceptions import (
     EntityAlreadyExistsException,
     EntityNotFoundException,
@@ -19,18 +20,21 @@
     UserId,
     UserPaginationParams,
 )
-from library.domains.interfaces.storages.user import IUserStorage
 
 
-class UserStorage(IUserStorage):
-    def __init__(self, session: AsyncSession) -> None:
-        self.session = session
+class UserStorage:
+    def __init__(self, *, uow: SqlalchemyUow) -> None:
+        self._uow = uow
+
+    @property
+    def _session(self) -> AsyncSession:
+        return self._uow.session
 
     async def fetch_user_by_id(self, *, user_id: UserId) -> User | None:
         stmt = select(UserTable).where(
             UserTable.id == user_id, UserTable.deleted_at.is_(None)
         )
-        user = (await self.session.scalars(stmt)).first()
+        user = (await self._session.scalars(stmt)).first()
         if user is None:
             return None
         return User(
@@ -45,15 +49,15 @@ async def exists_user_by_id(self, *, user_id: UserId) -> bool:
         stmt = select(
             exists().where(UserTable.id == user_id, UserTable.deleted_at.is_(None))
         )
-        return bool((await self.session.execute(stmt)).scalar())
+        return bool((await self._session.execute(stmt)).scalar())
 
     async def count_users(self, *, params: UserPaginationParams) -> int:
         query = (
             select(func.count())
             .select_from(UserTable)
             .where(UserTable.deleted_at.is_(None))
         )
-        result = (await self.session.execute(query)).scalar()
+        result = (await self._session.execute(query)).scalar()
         return result or 0
 
     async def fetch_user_list(self, *, params: UserPaginationParams) -> Sequence[User]:
@@ -69,7 +73,7 @@ async def fetch_user_list(self, *, params: UserPaginationParams) -> Sequence[Use
             .limit(params.limit)
             .offset(params.offset)
         )
-        result = (await self.session.execute(query)).mappings().all()
+        result = (await self._session.execute(query)).mappings().all()
         return [
             User(
                 id=UserId(result["id"]),
@@ -97,7 +101,7 @@ async def create_user(self, *, user: CreateUser) -> User:
             )
         )
         try:
-            result = (await self.session.execute(stmt)).mappings().one()
+            result = (await self._session.execute(stmt)).mappings().one()
         except IntegrityError as e:
             self._raise_error(e)
         return User(
@@ -114,7 +118,7 @@ async def delete_user_by_id(self, *, user_id: UserId) -> None:
             .where(UserTable.id == user_id)
             .values(deleted_at=datetime.now(tz=UTC))
         )
-        await self.session.execute(stmt)
+        await self._session.execute(stmt)
 
     async def update_user_by_id(self, *, update_user: UpdateUser) -> User:
         stmt = (
@@ -130,7 +134,7 @@ async def update_user_by_id(self, *, update_user: UpdateUser) -> User:
             )
         )
         try:
-            result = (await self.session.execute(stmt)).mappings().one()
+            result = (await self._session.execute(stmt)).mappings().one()
         except NoResultFound as e:
             raise EntityNotFoundException(entity=User, entity_id=update_user.id) from e
         return User(

library/adapters/database/uow.py

@@ -1,28 +1,54 @@
 import asyncio
+import logging
 from typing import Any
 
-from sqlalchemy.ext.asyncio import AsyncSession, AsyncSessionTransaction
+from sqlalchemy.ext.asyncio import (
+    AsyncSession,
+    AsyncSessionTransaction,
+    async_sessionmaker,
+)
 
 from library.domains.uow import AbstractUow
 
+logger = logging.getLogger(__name__)
+
 
 class SqlalchemyUow(AbstractUow):
-    transaction: AsyncSessionTransaction | None
+    _session: AsyncSession | None
+    _transaction: AsyncSessionTransaction | None
+    _session_factory: async_sessionmaker[AsyncSession]
+
+    def __init__(self, session_factory: async_sessionmaker[AsyncSession]) -> None:
+        self._session_factory = session_factory
+        self._session = None
+        self._transaction = None
 
-    def __init__(self, session: AsyncSession) -> None:
-        self.session = session
-        self.transaction = None
+    @property
+    def session(self) -> AsyncSession:
+        if self._session is None:
+            raise Exception("Session is not created")
+        return self._session
 
     async def commit(self) -> None:
         await self.session.commit()
 
     async def rollback(self) -> None:
         await self.session.rollback()
-        self.transaction = None
+        self._transaction = None
 
     async def create_transaction(self) -> None:
-        self.transaction = await self.session.begin()
+        if self._session is not None:
+            logger.warning("Attempt to create already existing session")
+            if self._transaction is not None:
+                raise Exception("Session is already in transaction")
+            else:
+                self._transaction = await self.session.begin()
+        else:
+            self._session = self._session_factory()
+            self._transaction = await self.session.begin()
 
     async def close_transaction(self, *exc: Any) -> None:
         task = asyncio.create_task(self.session.close())
         await asyncio.shield(task)
+        self._transaction = None
+        self._session = None