You cannot select more than 25 topics
Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.
447 lines
17 KiB
Python
447 lines
17 KiB
Python
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)
|