tpeetz/kontor
1
0
Fork 0
Code Issues 6 Pull Requests 2 Actions Packages Projects Releases 2 Wiki Activity

refactor scripts to work wit api

Browse Source
This commit is contained in:
Thomas Peetz
2025-04-16 05:08:59 +02:00
parent 98e3d91edd
commit 4a61d6a727
8 changed files with 262 additions and 191 deletions
Download Patch File Download Diff File Expand all files Collapse all files
fastapi/Makefile
+1 -1
View File
@@ -4,7 +4,7 @@ clean:
find . -name '*.py[co]' -delete
virtualenv:
virtualenv --prompt '|> kontor <| ' env
virtualenv --prompt '|> kontor-fastapi <| ' env
env/bin/pip install -r requirements.txt
env/bin/python setup.py develop
@echo
fastapi/app/main.py
-7
View File
@@ -1,5 +1,3 @@
from typing import Union
from fastapi import FastAPI
from .routers import comic, media
@@ -12,8 +10,3 @@ app.include_router(media.router)
@app.get("/")
def read_root():
return {"Hello": "World"}
@app.get("/items/{item_id}")
def read_item(item_id: int, q: Union[str, None] = None):
return {"item_id": item_id, "q": q}
fastapi/app/routers/comic.py
+13 -9
View File
@@ -1,10 +1,11 @@
from uuid import UUID
from fastapi import APIRouter
from fastapi import APIRouter, Depends, HTTPException
from sqlalchemy import select
from sqlalchemy.orm import Session
from app.models.comic import ComicResponse
from app.schema import Comic, __session__
from app.schema import Comic, get_db, SessionDep
router = APIRouter(
prefix="/comic",
@@ -14,15 +15,18 @@ router = APIRouter(
@router.get("/comics")
def get_all_comics() -> list[ComicResponse]:
def get_all_comics(db: SessionDep) -> list[ComicResponse]:
results: list[ComicResponse] = []
with __session__() as session:
comics = session.scalars(select(Comic)).all()
for comic in comics:
results.append(ComicResponse(id=comic.id, title=comic.title, completed=(comic.completed == 1)))
comics = db.scalars(select(Comic)).all()
for comic in comics:
results.append(ComicResponse(id=comic.id, title=comic.title, completed=(comic.completed == 1)))
return results
@router.get("/comics/{comic_id}")
def get_comic(comic_id: UUID) -> ComicResponse:
return ComicResponse(id=comic_id, title="Comic2", completed=False)
def get_comic(comic_id: UUID, db: SessionDep) -> ComicResponse:
comic = db.get(Comic, comic_id)
if comic is None:
raise HTTPException(status_code=404, detail="Comic could not be found")
response: ComicResponse = ComicResponse(id=comic_id, title=comic.title, completed=comic.completed)
return response
fastapi/app/routers/media.py
+64 -43
View File
@@ -1,12 +1,12 @@
from uuid import uuid4
from datetime import datetime
from typing import List
from uuid import uuid4, UUID
import mariadb
from fastapi import APIRouter, status, HTTPException
from fastapi.openapi.utils import status_code_ranges
from sqlalchemy import select
from sqlalchemy import select, Sequence
from app.models.media import MediaFileResponse, Link
from app.schema import MediaFile, __session__
from app.schema import MediaFile, SessionDep
router = APIRouter(
prefix="/media",
@@ -14,52 +14,73 @@ router = APIRouter(
)
@router.get("/update-titles")
def update_titles() -> list[MediaFileResponse]:
def update_titles(db: SessionDep) -> list[MediaFileResponse]:
results: list[MediaFileResponse] = []
with __session__() as session:
files = session.query(MediaFile).filter(MediaFile.review == 1).all()
for mediafile in files:
mediafile.update_title()
session.add(mediafile)
response = MediaFileResponse(id=mediafile.id,
title=mediafile.title,
file_name=mediafile.file_name,
cloud_link= mediafile.cloud_link,
url=str(mediafile.url),
review=(mediafile.review == 1),
should_download=(mediafile.should_download == 1))
results.append(response)
session.commit()
files = db.query(MediaFile).filter(MediaFile.review == 1).all()
for mediafile in files:
mediafile.update_title()
db.add(mediafile)
response = MediaFileResponse(id=mediafile.id,
title=mediafile.title,
file_name=mediafile.file_name,
cloud_link=mediafile.cloud_link,
url=str(mediafile.url),
review=(mediafile.review == 1),
should_download=(mediafile.should_download == 1))
results.append(response)
db.commit()
return results
@router.get("/files")
def get_files() -> list[MediaFileResponse]:
@router.get("/files", response_model=List[MediaFileResponse])
def get_files(db: SessionDep, review: bool = False, download: bool = False) -> List[MediaFileResponse]:
results: list[MediaFileResponse] = []
with __session__() as session:
files = session.scalars(select(MediaFile)).all()
for mediafile in files:
response = MediaFileResponse(id=mediafile.id,
title=mediafile.title,
file_name=mediafile.file_name,
cloud_link= mediafile.cloud_link,
url=str(mediafile.url),
review=(mediafile.review == 1),
should_download=(mediafile.should_download == 1))
results.append(response)
files: Sequence[MediaFile]
if review:
files = db.query(MediaFile).filter(MediaFile.review == 1).all()
elif download:
files = db.query(MediaFile).filter(MediaFile.should_download == 1).all()
else:
files = db.scalars(select(MediaFile)).all()
for mediafile in files:
response = MediaFileResponse(id=mediafile.id,
title=mediafile.title,
file_name=mediafile.file_name,
cloud_link=mediafile.cloud_link,
url=str(mediafile.url),
review=(mediafile.review == 1),
should_download=(mediafile.should_download == 1))
results.append(response)
return results
@router.put("/files/{file_id}", response_model=MediaFileResponse)
def update_file(file_id: UUID, db: SessionDep, info: MediaFileResponse) -> MediaFileResponse:
mediaFile = db.get(MediaFile, file_id)
if not mediaFile:
raise HTTPException(status_code=404, detail="MediaFile could not be found")
mediaFile.file_name = info.file_name
mediaFile.cloud_link = info.cloud_link
mediaFile.url = info.url
mediaFile.title = info.title
mediaFile.last_modified_date = datetime.now()
mediaFile.review = info.review
mediaFile.should_download = info.should_download
db.add(mediaFile)
db.commit()
return info
@router.post("/files", status_code=status.HTTP_201_CREATED)
def add_file(new_link: Link) -> MediaFileResponse:
def add_file(new_link: Link, db: SessionDep) -> MediaFileResponse:
print(new_link.url)
try:
with __session__() as session:
mediaFile: MediaFile = MediaFile()
setattr(mediaFile, "url", new_link.url)
setattr(mediaFile, "review", True)
setattr(mediaFile, "should_download", True)
session.add(mediaFile)
session.commit()
except :
mediaFile: MediaFile = MediaFile()
setattr(mediaFile, "url", new_link.url)
setattr(mediaFile, "review", True)
setattr(mediaFile, "should_download", True)
db.add(mediaFile)
db.commit()
except:
raise HTTPException(status_code=409, detail="Link duplicate")
response = MediaFileResponse(id=uuid4(),
title=mediaFile.title,
@@ -67,5 +88,5 @@ def add_file(new_link: Link) -> MediaFileResponse:
cloud_link=mediaFile.cloud_link,
url=new_link.url,
review=(mediaFile.review == 1),
shoud_download=(mediaFile.should_download==1))
should_download=(mediaFile.should_download == 1))
return response
fastapi/app/schema/__init__.py
+12 -3
View File
@@ -1,10 +1,12 @@
import logging
import logging.config
from pathlib import Path
from typing import Annotated
import yaml
from fastapi import Depends
from platformdirs import PlatformDirs
from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import sessionmaker, Session
from .admin import User, Token, Role, AuthorizationMatrix, ModuleData, MailAccount, Mail
from .bookshelf import Article, Book, Author, BookshelfPublisher, ArticleAuthor, BookAuthor
@@ -34,5 +36,12 @@ connect_string = ('mariadb+mariadbconnector://{}:{}@{}:{}/{}'.format(
db_config['mariadb']['database']
))
engine = create_engine(connect_string)
SessionLocal = sessionmaker(bind=engine)
Base.metadata.create_all(bind=engine, checkfirst=True)
__session__ = sessionmaker(engine)
def get_db():
logger.info("get_db")
with SessionLocal() as db:
yield db
SessionDep = Annotated[Session, Depends(get_db)]