the architecture of repositories and UoF is fully implemented, tests …

audiostats/application/dto_mappers.py

@@ -0,0 +1,47 @@
+from audiostats.db import Album, Track
+
+from audiostats.handlers import AlbumDTO, TrackDTO
+
+
+# def create_album_f_dto(album : AlbumDTO) -> Album:
+#     return Album(title=album.title,
+#                 performer=album.performer,
+#                 year=album.year,
+#                 path=album.path,
+#                 cover=album.cover,
+#                 tracks=[Track(title=track.title,
+#                               number=track.number,
+#                               path=track.path,
+#                               offset=track.offset,
+#                               duration=track.duration) for track in album.tracks])
+
+def update_track_orm_f_dto(old : Track, new : TrackDTO) -> None:
+    old.title = new.title
+    old.number = new.number
+    old.path = new.path
+    old.offset = new.offset
+    old.duration = new.duration
+
+def create_track_orm_f_dto(track : TrackDTO) -> Track:
+    created = Track()
+    update_track_orm_f_dto(created, track)
+    return created
+
+def update_album_orm_meta_f_dto(old : Album, new : AlbumDTO):
+    old.title = new.title
+    old.performer = new.performer
+    old.year = new.year
+    old.path = new.path
+    old.cover = new.cover
+
+def create_album_dto_f_orm(album : Album):
+    return AlbumDTO(title=album.title,
+                    performer=album.performer,
+                    year=album.year,
+                    path=album.path,
+                    cover=album.cover,
+                    tracks=[TrackDTO(title=track.title,
+                                     number=track.number,
+                                     path=track.path,
+                                     offset=track.offset,
+                                     duration=track.duration) for track in album.tracks])

audiostats/db/__init__.py

@@ -0,0 +1,2 @@
+from .models import Album, Track
+from .api import DBApi

audiostats/db/api.py

@@ -1,6 +1,31 @@
+import logging
+import time
+from collections.abc import Iterator
 
+from sqlalchemy.orm import sessionmaker, Session
+from .uof import UnitOfWork
 
+from audiostats.handlers import AlbumDTO
 
 
+logger = logging.getLogger(__name__)
+
 class DBApi:
-    ...
+    def __init__(self, session_factory : sessionmaker[Session]):
+        self._session_factory = session_factory
+
+    def upsert_albums(self, albums : Iterator[AlbumDTO]):
+        t_start = time.time()
+        with UnitOfWork(self._session_factory) as uof:
+            total, success = 0, 0
+            for album in albums:
+                uof.albums.upsert(album)
+                total += 1
+                success += 1
+            logger.info(f'Albums updated/inserted in {(time.time() - t_start) * 1000:.3f} ms. Total: {total}, success: {success}')
+
+    def get_all_albums(self):
+        t_start = time.time()
+        with UnitOfWork(self._session_factory) as uof:
+            return uof.albums.all()
+

audiostats/db/repositories.py

@@ -0,0 +1,38 @@
+from sqlalchemy import select
+from sqlalchemy.orm import Session
+
+from .models import Album
+from audiostats.handlers import AlbumDTO
+
+from audiostats.application.dto_mappers import create_album_dto_f_orm, update_album_orm_meta_f_dto, update_track_orm_f_dto, create_track_orm_f_dto
+
+
+class AlbumRepository:
+    def __init__(self, session : Session):
+        self._session = session
+
+    def upsert(self, album_data : AlbumDTO):
+        album = self.find_by_title_performer(album_data.title, album_data.performer)
+
+        if not album:
+            album = Album()
+            self._session.add(album)
+        update_album_orm_meta_f_dto(album, album_data)
+
+        old_tracks_by_title = {track.title: track for track in album.tracks}
+        for dto in album_data.tracks:
+            if dto.title in old_tracks_by_title:
+                old_track = old_tracks_by_title.pop(dto.title)
+                update_track_orm_f_dto(old_track, dto)
+            else:
+                track = create_track_orm_f_dto(dto)
+                album.tracks.append(track)
+
+        for track in old_tracks_by_title.values():
+            album.tracks.remove(track)
+
+    def find_by_title_performer(self, title : str, performer : str | None) -> Album | None:
+        return self._session.query(Album).filter(Album.title == title and Album.performer == performer if performer else Album.performer.is_(None)).first()
+
+    def all(self) -> list[Album]:
+        return [create_album_dto_f_orm(album) for album in self._session.scalars(select(Album)).all()]

audiostats/db/uof.py

@@ -0,0 +1,28 @@
+from sqlalchemy.orm import sessionmaker, Session
+import logging
+
+from .repositories import AlbumRepository
+
+logger = logging.getLogger(__name__)
+
+class UnitOfWork:
+    def __init__(self, session_factory: sessionmaker[Session]):
+        self._session_factory = session_factory
+        self._session : Session | None = None
+        self.albums : AlbumRepository | None = None
+        logger.debug(f'UoF initialized: {self}')
+
+    def __enter__(self):
+        self._session = self._session_factory()
+        self.albums = AlbumRepository(self._session)
+        logger.debug(f'AlbumRepo initialised: {self.albums}, Current Session: {self._session}')
+        return self
+
+    def __exit__(self, exc_type, exc_val, exc_tb):
+        if exc_type is not None:
+            self._session.rollback()
+        else:
+            self._session.commit()
+        self._session.close()
+        logger.debug(f'Session closed')
+

tests/__init__.py

@@ -0,0 +1 @@
+from .db_fixture import test_engine, test_sessionmaker

tests/db_fixture.py

@@ -0,0 +1,22 @@
+import logging
+
+import pytest
+from sqlalchemy import create_engine
+from sqlalchemy.orm import sessionmaker
+
+from audiostats.db.models import Base
+
+logger = logging.getLogger(__name__)
+
+@pytest.fixture(scope="session")
+def test_engine():
+    engine = create_engine("sqlite:///:memory:", connect_args={"check_same_thread": False})
+    Base.metadata.create_all(engine)
+    yield engine
+    engine.dispose()
+
+@pytest.fixture(scope="session")
+def test_sessionmaker(test_engine):
+    sessionfactory = sessionmaker(bind=test_engine)
+    logger.debug(f'session type: {type(sessionfactory)}')
+    return sessionfactory

tests/test_db_api.py

@@ -0,0 +1,18 @@
+import logging
+
+from audiostats.db.api import DBApi
+from tests import test_sessionmaker, test_engine
+
+def test_upsert_albums(test_sessionmaker, processed_album_dtos):
+    api = DBApi(test_sessionmaker)
+
+    api.upsert_albums(processed_album_dtos)
+
+    albums = api.get_all_albums()
+
+    assert albums.sort(key=lambda x: x.title) == processed_album_dtos.sort(key=lambda x: x.title)
+
+
+
+
+