extract SQLAlchemy model in separate python library

kontor-api/src/db/repository/comics/artist.py

@@ -1,47 +0,0 @@
-import uuid
-from typing import List, Optional
-
-from sqlalchemy.orm import Session
-
-from src.core.log_conf import logger
-from src.db.models.comic import Artist
-from src.schema.comics.artist import AddArtist
-from src.schema.comics.artist_details import ArtistDetailResponse, ArtistWorktypeComicResponse, ArtistWorktypeIssueResponse
-from src.schema.comics.comic import ComicResponse
-from src.schema.comics.worktype import WorktypeResponse
-
-
-def get_artist_details(artist: Artist) -> ArtistDetailResponse:
-    comic_works: List[ArtistWorktypeComicResponse] = []
-    comic_works_map = {}
-    for work in artist.comic_works:
-        worktype_id = work.work_type.id
-        if worktype_id in comic_works_map:
-            comic = ComicResponse(id=work.comic.id, title=work.comic.title, completed=work.comic.completed)
-            comic_works_map[worktype_id].comics.append(comic)
-        else:
-            comic_works_map[worktype_id] = ArtistWorktypeComicResponse(
-                worktype=WorktypeResponse(id=worktype_id, name=work.work_type.name),
-                comics=[ComicResponse(id=work.comic.id, title=work.comic.title, completed=work.comic.completed)]
-            )
-    for value in comic_works_map.values():
-        comic_works.append(value)
-    issue_works: List[ArtistWorktypeIssueResponse] = []
-    response = ArtistDetailResponse(
-        id=artist.id,
-        name=str(artist.name),
-        weblink=str(artist.weblink),
-        comic_works=comic_works,
-        issue_works=issue_works,
-    )
-    return response
-
-def update_artist(add_artist: AddArtist, artist_id: str, db: Session) -> Optional[Artist]:
-    logger.info("update artist")
-    artist: Optional[Artist] = db.get(Artist, artist_id)
-    if artist is not None:
-        artist.name = add_artist.name
-        db.add(artist)
-        db.commit()
-        db.refresh(artist)
-    return artist

kontor-api/src/db/repository/comics/comic.py

@@ -1,72 +0,0 @@
-from typing import Dict, List, Optional
-
-from sqlalchemy.orm import Session
-
-from src.core.log_conf import logger
-from src.db.models.comic import Comic, Issue
-from src.schema.comics.artist import artist_to_response
-from src.schema.comics.comic import ComicSchema, comic_to_response
-from src.schema.comics.comic_details import ComicDetailsResponse, ComicWorktypeArtistResponse
-from src.schema.comics.issue import IssueResponse, issue_to_response
-from src.schema.comics.issue_details import IssueDetailsResponse
-from src.schema.comics.publisher import publisher_to_response
-from src.schema.comics.volume import VolumeResponse, volume_to_response
-from src.schema.comics.worktype import worktype_to_response
-
-
-def get_issue_details(issue: Issue) -> IssueDetailsResponse:
-    volume = None
-    if issue.volume:
-        volume = volume_to_response(issue.volume)
-    response = IssueDetailsResponse(
-        id=issue.id,
-        issue_number=str(issue.issue_number),
-        in_stock=bool(issue.in_stock),
-        is_read=bool(issue.is_read),
-        comic=comic_to_response(issue.comic),
-        volume=volume
-    )
-    return response
-
-
-def update_comic(new_comic: ComicSchema, comic_id: str, db: Session) -> Optional[Comic]:
-    logger.info(f"update_comic: {new_comic} with {comic_id}")
-    comic: Optional[Comic] = db.get(Comic, comic_id)
-    return comic
-
-def get_comic_details(comic: Comic) -> ComicDetailsResponse:
-    volumes: List[VolumeResponse] = []
-    for volume in comic.volumes:
-        volumes.append(volume_to_response(volume))
-    issues: List[IssueResponse] = []
-    for issue in comic.issues:
-        issues.append(issue_to_response(issue))
-    works: List[ComicWorktypeArtistResponse] = []
-    works_map: Dict[str, ComicWorktypeArtistResponse] = {}
-    for work in comic.comic_works:
-        worktype_id = work.work_type.id
-        if worktype_id in works_map:
-            artist = artist_to_response(work.artist)
-            works_map[worktype_id].artists.append(artist)
-            logger.info(f"add artist to response map: {artist} -> {works_map}")
-            print(f"add artist to response map: {artist} -> {works_map}")
-        else:
-            works_map[worktype_id] = ComicWorktypeArtistResponse(
-                worktype=worktype_to_response(work.work_type),
-                artists=[artist_to_response(work.artist)]
-            )
-    for value in works_map.values():
-        works.append(value)
-    response = ComicDetailsResponse(
-        id=str(comic.id),
-        created=str(comic.created_date),
-        title=str(comic.title),
-        completed=bool(comic.completed),
-        current_order=bool(comic.current_order),
-        weblink=str(comic.weblink),
-        publisher=publisher_to_response(comic.publisher),
-        issues=issues,
-        volumes=volumes,
-        works=works
-    )
-    return response

kontor-api/src/db/repository/comics/publisher.py

@@ -1,24 +0,0 @@
-from typing import List
-
-from src.db.models.comic import Publisher
-from src.schema.comics.comic import ComicResponse
-from src.schema.comics.publisher import PublisherResponse
-from src.schema.comics.publisher_details import PublisherDetailsResponse
-
-
-def get_publisher_details(publisher: Publisher) -> PublisherDetailsResponse:
-    imprints: List[PublisherResponse] = []
-    for imprint in publisher.imprints:
-        imprints.append(PublisherResponse(id=imprint.id, name=str(imprint.name)))
-    comics: List[ComicResponse] = []
-    for comic in publisher.comics:
-        comics.append(
-            ComicResponse(id=comic.id, title=comic.title, completed=comic.completed)
-        )
-    parent_publisher: PublisherResponse | None = None
-    if publisher.parent_publisher:
-        parent_publisher = PublisherResponse(id=publisher.parent_publisher.id, name=str(publisher.parent_publisher.name))
-    response: PublisherDetailsResponse = PublisherDetailsResponse(
-        id=publisher.id, name=str(publisher.name), parent_publisher=parent_publisher, imprints=imprints, comics=comics
-    )
-    return response

kontor-api/src/db/repository/comics/worktype.py

@@ -1,33 +0,0 @@
-import uuid
-from datetime import datetime
-from typing import AnyStr, Optional
-
-from sqlalchemy.orm import Session
-
-from src.core.log_conf import logger
-from src.db.models.comic import WorkType
-from src.schema.comics.worktype import AddWorkType
-
-
-def create_new_worktype(work: AddWorkType, db: Session) -> WorkType:
-    worktype = WorkType()
-    worktype.id = str(uuid.uuid4())
-    worktype.created_date = datetime.now()
-    worktype.last_modified_date = datetime.now()
-    worktype.name = work.worktype
-    db.add(worktype)
-    db.commit()
-    db.refresh(worktype)
-    logger.info(f"create_new_worktype: {worktype}")
-    return worktype
-
-
-def update_worktype(work: AddWorkType, worktype_id: AnyStr, db: Session) -> Optional[WorkType]:
-    logger.info("update worktype")
-    worktype: Optional[WorkType] = db.get(WorkType, worktype_id)
-    if worktype is not None:
-        worktype.name = work.worktype
-        db.add(worktype)
-        db.commit()
-        db.refresh(worktype)
-    return worktype