Vorbereitung Release 0.2.0
This commit is contained in:
@@ -0,0 +1,46 @@
|
||||
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) -> Artist:
|
||||
logger.info("update artist")
|
||||
artist: Optional[Artist] = db.get(Artist, artist_id)
|
||||
artist.name = add_artist.name
|
||||
db.add(artist)
|
||||
db.commit()
|
||||
db.refresh(artist)
|
||||
return artist
|
||||
@@ -0,0 +1,87 @@
|
||||
from typing import Dict, List
|
||||
|
||||
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 ArtistResponse
|
||||
from src.schema.comics.comic import ComicResponse, ComicSchema
|
||||
from src.schema.comics.comic_details import ComicDetailsResponse, ComicWorktypeArtistResponse
|
||||
from src.schema.comics.issue import IssueResponse
|
||||
from src.schema.comics.issue_details import IssueDetailsResponse
|
||||
from src.schema.comics.publisher import PublisherResponse
|
||||
from src.schema.comics.volume import VolumeResponse
|
||||
from src.schema.comics.worktype import WorktypeResponse
|
||||
|
||||
|
||||
def list_comics(db: Session) -> List[Comic]:
|
||||
comics = db.query(Comic).all()
|
||||
return comics
|
||||
|
||||
|
||||
def get_issue_details(issue: Issue) -> IssueDetailsResponse:
|
||||
response = IssueDetailsResponse(
|
||||
id=issue.id,
|
||||
issue_number=str(issue.issue_number),
|
||||
in_stock=bool(issue.in_stock),
|
||||
is_read=bool(issue.is_read),
|
||||
comic=ComicResponse(id=issue.comic.id, title=issue.comic.title, completed=issue.comic.completed),
|
||||
volume=VolumeResponse(id=issue.volume.id, name=issue.volume.name)
|
||||
)
|
||||
return response
|
||||
|
||||
|
||||
def update_comic(comic: ComicSchema, comic_id: str, db: Session) -> type[Comic] | None:
|
||||
logger.info(f"update_comic: {comic} with {comic_id}")
|
||||
comic = db.get(Comic, comic_id) # type: ignore
|
||||
return comic # type: ignore
|
||||
|
||||
def get_short_info(comic: Comic) -> ComicResponse:
|
||||
response = ComicResponse(
|
||||
id=comic.id,
|
||||
title=str(comic.title),
|
||||
completed=bool(comic.completed == 1)
|
||||
)
|
||||
return response
|
||||
|
||||
def get_comic_details(comic: Comic) -> ComicDetailsResponse:
|
||||
volumes: List[VolumeResponse] = []
|
||||
for volume in comic.volumes:
|
||||
volumes.append(VolumeResponse(id=volume.id, name=volume.name))
|
||||
issues: List[IssueResponse] = []
|
||||
for issue in comic.issues:
|
||||
issues.append(IssueResponse(
|
||||
id=issue.id,
|
||||
issue_number=issue.issue_number,
|
||||
in_stock=issue.in_stock,
|
||||
is_read=issue.is_read
|
||||
))
|
||||
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 = ArtistResponse(id=work.artist.id, name=work.artist.name)
|
||||
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=WorktypeResponse(id=worktype_id, name=work.work_type.name),
|
||||
artists=[ArtistResponse(id=work.artist.id, name=work.artist.name)]
|
||||
)
|
||||
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=PublisherResponse(id=comic.publisher.id, name=comic.publisher.name),
|
||||
issues=issues,
|
||||
volumes=volumes,
|
||||
works=works
|
||||
)
|
||||
return response
|
||||
@@ -0,0 +1,24 @@
|
||||
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
|
||||
@@ -0,0 +1,32 @@
|
||||
import uuid
|
||||
from datetime import datetime
|
||||
from typing import AnyStr
|
||||
|
||||
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) -> WorkType:
|
||||
logger.info("update worktype")
|
||||
worktype = db.get(WorkType, worktype_id)
|
||||
worktype.name = work.worktype
|
||||
db.add(worktype)
|
||||
db.commit()
|
||||
db.refresh(worktype)
|
||||
return worktype
|
||||
Reference in New Issue
Block a user