from urllib.parse import unquote from fastapi import FastAPI, HTTPException import uvicorn from typing import List, Tuple from database.database import Database from database.database import DBPresence, DBArt2Presence, DBCollection2Art from database.models import Art, ArtnoID, Presence, ArtistNoId, Tag, TagNoID, TagCategory, TagCategorynoID, TopicNoId, \ Topic, CollectionNoID, Collection, Art2CollRelationNoID, Art2CollRelation app = FastAPI() db = Database() @app.get("/artnet/metadata/art") async def art(id: int = None, hash: str = None, tag_id: int = None, collection_id: int = None): hash = unquote(hash) if hash is not None else None print(f"Received GET on /artnet/metadata/art (id={id}, hash={hash}, tag_id={tag_id})") result = None if id is not None: result = db.get_art_by_id(id) elif hash is not None: result = db.get_art_by_hash(hash) elif tag_id is not None: result = db.get_tag_art(tag_id) elif collection_id is not None: result = db.get_collection_art(collection_id) if id is None and hash is None and tag_id is None: # TODO fix against listing enormous amounts of art (because "all" could be large) result = db.get_art_list() if result is not None: return result raise HTTPException(status_code=404, detail="Art was not found!") @app.post("/artnet/metadata/art") async def art(art: ArtnoID, id: int = None): print(f"Received POST on /artnet/metadata/art (id={id}) body: art=({art})") if id is None: # create new art entry if art.presences is None: # tried to create art without any presence raise HTTPException(status_code=422, detail="No presences were listed") try: new_id = db.create_art_by_model(art).id return {"id": new_id} except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) else: # update existing art entry if db.get_art_by_id(id) is None: raise HTTPException(status_code=404, detail="The specified art could not be found!") updated_art = Art(id=id) updated_art.hash = art.hash updated_art.link = art.link updated_art.title = art.title updated_art.path = art.path updated_art.description = art.description if art.presences is not None: updated_art.presences = art.presences if art.collections is not None: updated_art.collections = art.collections db.update_art_by_model(updated_art) return True @app.delete("/artnet/metadata/art") async def art(id: int, presence_name: str = None, presence_domain: str = None): presence_name = unquote(presence_name) if presence_name is not None else None presence_domain = unquote(presence_domain) if presence_domain is not None else None print(f"Received DELETE on /artnet/metadata/art (id={id}, presence_name={presence_name}, " f"presence_domain={presence_domain})") print("All art available is", [a.id for a in db.get_art_list()]) if db.get_art_by_id(id) is None: raise HTTPException(status_code=404, detail="Art has not been found!") if id is not None and presence_name is None and presence_domain is None: db.delete_art_by_id(id) return elif id is not None and presence_name is not None and presence_domain is not None: try: db.delete_art_presences_by_id(art_id=id, presence_name=presence_name, presence_domain=presence_domain) except ValueError: raise HTTPException(status_code=404, detail="The art-presence relation could not be found!") return raise HTTPException(status_code=422, detail="Unknown parameter combination!") @app.get("/artnet/metadata/presence") async def presence(name: str = None, domain: str = None, artist_id: int = None, art_id: int = None, art_md5: str = None): name = unquote(name) if name is not None else None domain = unquote(domain) if domain is not None else None art_md5 = unquote(art_md5) if art_md5 is not None else None print(f"Received GET on /artnet/metadata/presence (name={name}, domain={domain}, artist_id={artist_id}" f", art_id={art_id}, art_md5={art_md5})") result = None if artist_id is not None and name is None and domain is None: result = db.get_artist_presences(artist_id) elif art_id is not None and name is None and domain is None: result = db.get_art_presences_by_id(art_id) elif art_md5 is not None and name is None and domain is None: result = db.get_art_presences_by_hash(art_md5) elif name is not None and domain is not None: result = db.get_presence(name, domain) if name is None and domain is None and artist_id is None and art_id is None and art_md5 is None: result = db.get_presence_list() #raise HTTPException(status_code=406, detail="You must query with at least one parameter!") if result is not None: if isinstance(result, list): for i in range(len(result)): if isinstance(result[i], DBPresence): result[i].name = result[i].name.strip() result[i].domain = result[i].domain.strip() elif isinstance(result[i], DBArt2Presence): result[i].presence_name = result[i].presence_name.strip() result[i].presence_domain = result[i].presence_domain.strip() else: result.name = result.name.strip() result.domain = result.domain.strip() return result raise HTTPException(status_code=404, detail="Presence was not found!") @app.post("/artnet/metadata/presence") async def presence(presence: Presence): print(f"Received POST on /artnet/metadata/presence body: presence={presence}") curr_presence = db.get_presence(name=presence.name, domain=presence.domain) r = db.get_artist(presence.artist_id) if r is None: raise HTTPException(status_code=406, detail="Invalid Artist ID! Could not find artist.") if curr_presence is None: # must be new presence db.create_presence(name=presence.name, domain=presence.domain, artist_id=presence.artist_id, link=presence.link) else: # editing current presence db.update_presence(name=curr_presence.name, domain=curr_presence.domain, artist_id=presence.artist_id, link=presence.link) @app.delete("/artnet/metadata/presence") async def presence(name: str, domain: str): print(f"Received DELETE on /artnet/metadata/presence (name={name}, domain={domain})") if db.get_presence(name, domain) is None: raise HTTPException(status_code=404, detail="Presence has not been found!") db.delete_presence(name, domain) @app.get("/artnet/metadata/artist") async def artist(id: int = None, topic_id: int = None, name: str = None): name = unquote(name) if name is not None else None print(f"Received GET on /artnet/metadata/artist (id={id}, topic_id={topic_id}, name={name})") result = None if id is None and topic_id is None and name is None: result = db.get_artist_list() if id is not None: result = db.get_artist(id) elif topic_id is not None: result = db.get_topic_artists(topic_id) elif name is not None: result = db.search_artist(name) if result is not None: return result raise HTTPException(status_code=404, detail="Artist was not found!") @app.post("/artnet/metadata/artist") async def artist(artist: ArtistNoId, id: int = None): print(f"Received POST on /artnet/metadata/artist (id={id}) body: artist={artist}") if id is None: # create new artist db_artist = db.create_artist(name=artist.name, topics=artist.topics) return db_artist.id else: if db.get_artist(id) is not None: db.update_artist(artist_id=id, name=artist.name, topics=artist.topics) else: raise HTTPException(status_code=404, detail="Tried to edit unknown artist. ID was not found!") @app.delete("/artnet/metadata/artist") async def artist(id: int): print(f"Received DELETE on /artnet/metadata/artist (id={id})") if db.get_artist(id) is None: raise HTTPException(status_code=404, detail="Tried to delete unknown artist. ID was not found!") db.delete_artist(id) @app.get("/artnet/metadata/topic") async def topic(name: str = None, id: int = None, artist_id: int = None): name = unquote(name) if name is not None else None print(f"Received GET on /artnet/metadata/topic (name={name}, id={id}, artist_id={artist_id})") result = None if name is None and id is None and artist_id is None: result = db.get_topic_list() if name is not None or id is not None: if id is not None: result = db.get_topic_by_id(id) elif name is not None: result = db.get_topic_by_name(name) elif artist_id is not None: result = db.get_artist_topics(artist_id) if result is not None: if isinstance(result, list): for i in range(len(result)): result[i].name = result[i].name.strip() else: result.name = result.name.strip() return result raise HTTPException(status_code=404, detail="Topic was not found!") @app.post("/artnet/metadata/topic") async def topic(topic: TopicNoId, id: int = None): print(f"Received POST on /artnet/metadata/topic (id={id}) body: topic={topic}") if id is None: # creating new topic try: new_id = db.create_topic_by_model(topic).id return {"id": new_id} except ValueError as e: raise HTTPException(status_code=406, detail=str(e)) else: # updating topic topic_ = Topic(name=topic.name, description=topic.description, id=id) db.update_topic_by_model(topic_) @app.delete("/artnet/metadata/topic") async def topic(id: int): print(f"Received DELETE on /artnet/metadata/topic (id={id})") if db.get_topic_by_id(id) is None: raise HTTPException(status_code=404, detail="No topic with this id could be found!") db.delete_topic_by_id(id) @app.get("/artnet/metadata/tag") async def tag(id: int = None, name: str = None, fuzzy: bool = False, category: int = None): name = unquote(name) if name is not None else None print(f"Received GET on /artnet/metadata/tag (name={name}, id={id}, name={category}, fuzzy={fuzzy})") result = None if id is None and name is None and category is None: result = db.get_tag_list() if fuzzy: if name is not None: result = db.search_tag_by_name_fuzzy(name) if result is not None: return result else: if id is not None: result = db.get_tag_by_id(id) elif name is not None: result = db.get_tag_by_name(name) elif category is not None: result = db.get_category_tags(category) if result is not None: if isinstance(result, list): for i in range(len(result)): result[i].name = result[i].name.strip() else: result.name = result.name.strip() return result raise HTTPException(status_code=404, detail="No matching tag found!") @app.post("/artnet/metadata/tag") async def tag(tag: TagNoID, id: int = None): print(f"Received POST on /artnet/metadata/tag (id={id}) body: tag={tag}") if id is None: # create new tag tag_id = db.create_tag_by_model(tag).id return {"id": tag_id} else: # update already existing tag tag = Tag(id=id, name=tag.name, description=tag.description, category_id=tag.category_id) try: db.update_tag_by_model(tag) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) return True @app.delete("/artnet/metadata/tag") async def tag(id: int): print(f"Received DELETE on /artnet/metadata/tag (id={id})") try: db.delete_tag_by_id(tag_id=id) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) @app.get("/artnet/metadata/category") async def category(name: str = None, id: int = None): name = unquote(name) if name is not None else None print(f"Received GET on /artnet/metadata/category (name={name}, id={id})") if name is None and id is None: result = db.get_category_list() return result result = None if id is not None: result = db.get_category_by_id(id) elif name is not None: result = db.get_category_by_name(name) if result is not None: result.name = result.name.strip() return result raise HTTPException(status_code=404, detail="Category not found!") @app.post("/artnet/metadata/category") async def category(category: TagCategorynoID, id: int = None): print(f"Received GET on /artnet/metadata/category (id={id}, body: {category})") if id is None: # create new name try: new_id = db.create_category_by_model(category).category_id return {"id": new_id} except ValueError as e: raise HTTPException(status_code=422, detail=str(e)) else: updated_category = TagCategory(category_id=id, name=category.name) db.update_category_by_model(updated_category) return True @app.delete("/artnet/metadata/category") def category(id: int): print(f"Received DELETE on /artnet/metadata/category (id={id})") try: db.delete_category_by_id(category_id=id) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) @app.get("/artnet/metadata/collection") def collection(id: int = None, name: str = None, art_id: int = None): print(f"Received GET on /artnet/metadata/collection (id={id}, name={name})") result = None if id is None and name is None and art_id is None: result = db.get_collection_list() if id is not None and name is not None: raise HTTPException(status_code=406, detail="Both id and name are not supported!") if id is not None: result = db.get_collection_by_id(id) if name is not None: result = db.get_collections_by_name(name) if art_id is not None: result = db.get_art_collections(art_id) if result is not None: if isinstance(result, list): for i in range(len(result)): if isinstance(result[i], DBCollection2Art): continue result[i].name = result[i].name.strip() else: result.name = result.name.strip() return result @app.post("/artnet/metadata/collection") async def collection(collection: CollectionNoID, id: int = None): print(f"Received POST on /artnet/metadata/collection (id={id})") if id is None: # create new collection new_id = db.create_collection_by_model(collection).id return {"id": new_id} else: updated_collection = Collection(id=id, name=collection.name, description=collection.description) db.update_collection_by_model(updated_collection) return True @app.delete("/artnet/metadata/collection") async def collection(id: int): print(f"Received DELETE on /artnet/metadata/collection (id={id})") try: db.delete_collection_by_id(id) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) @app.get("/artnet/metadata/collection_entry") async def collection_entry(art_id: int, collection_id: int): print(f"Received GET on /artnet/metadata/collection_entry (art_id={art_id}, collection_id={collection_id})") result = db.get_art_collection_relation(art_id=art_id, collection_id=collection_id) if result is None: raise HTTPException(status_code=404, detail="Can not find a art-collection-relation with these IDs!") return result @app.post("/artnet/metadata/collection_entry") async def collection_entry(art2collection: Art2CollRelationNoID, art_id: int = None, collection_id: int = None): print(f"Received POST on /artnet/metadata/collection_entry") if art_id is None and collection_id is None: # create new relation, impossible! raise HTTPException(status_code=406, detail="You can not create a relation without IDs!") elif art_id is not None and collection_id is not None: art2collection_id = Art2CollRelation(art_id=art_id, collection_id=collection_id) art2collection_id.ranking = art2collection.ranking db.update_art_collection_relation_by_model(art2collection_id) return raise HTTPException(status_code=406, detail="You can not only specify only one ID!") @app.delete("/artnet/metadata/collection_entry") async def collection_entry(art_id: int, collection_id: int): print(f"Received DELETE on /artnet/metadata/collection_entry (art_id={art_id}, collection_id={collection_id})") try: db.delete_art_collection_relation(art_id=art_id, collection_id=collection_id) except ValueError as e: raise HTTPException(status_code=404, detail=str(e)) if __name__ == "__main__": uvicorn.run(app, host="127.0.0.1", port=8000)