diff --git a/database/database.py b/database/database.py index 6d79d69..fb6b948 100644 --- a/database/database.py +++ b/database/database.py @@ -2,10 +2,11 @@ from typing import List import psycopg2 from sqlalchemy import Column, Boolean, Float, String, Integer, ForeignKey, Table, func, ForeignKeyConstraint +import sqlalchemy.orm as db import sqlalchemy from sqlalchemy.orm import declarative_base, relationship, load_only -from database.models import Art, ArtnoID, Artist, Presence, TagCategory, TagCategorynoID +from database.models import Art, ArtnoID, Artist, Presence, TagCategory, TagCategorynoID, TagNoID, Tag # TODO read sensitive data from environment file or similar SQLALCHEMY_DATABASE_URL = "postgresql://artnet_editor:G606Rm9sFEXe6wfTLxVu@[::1]/test_artnet" @@ -28,6 +29,20 @@ artist_to_topic_table = Table('artist_to_topic', Base.metadata, Column('topic_id', Integer, ForeignKey('topic.id'))) +class DBTagImplication(Base): + __tablename__ = 'tag_implication' + + root_tag = Column(Integer, ForeignKey('tag.tag_id'), primary_key=True) + implicate = Column(Integer, ForeignKey('tag.tag_id'), primary_key=True) + + +class DBTagAlias(Base): + __tablename__ = 'tag_alias' + + tag1 = Column(Integer, ForeignKey('tag.tag_id'), primary_key=True) + tag2 = Column(Integer, ForeignKey('tag.tag_id'), primary_key=True) + + class DBPresence(Base): __tablename__ = "presence" @@ -122,6 +137,10 @@ class DBTag(Base): # TODO check if cascade is required art = relationship("DBArt", secondary=art_to_tag_table, back_populates="tags") # TODO check if cascade is required + implications = relationship(DBTagImplication, backref="implied_by", + primaryjoin=tag_id == DBTagImplication.root_tag) + alias = relationship(DBTagAlias, backref="alias", + primaryjoin=(tag_id == DBTagAlias.tag1) or (tag_id == DBTagAlias.tag2)) Base.metadata.create_all(bind=engine) @@ -382,6 +401,9 @@ class Database: # Tag + def get_tag_list(self): + return self.__get_db().query(DBTag).all() # TODO fix StackOverflow + def get_tag_by_id(self, tag_id: int) -> DBTag: result = self.__get_db().query(DBTag).where(tag_id == DBTag.tag_id).first() return result @@ -390,14 +412,66 @@ class Database: result = self.__get_db().query(DBTag).filter(func.lower(DBTag.name) == name.lower()).first() return result - def update_tag(self, tag_id: int, name: str, description: str, category: str): - raise NotImplementedError + def create_tag_by_model(self, tag: TagNoID): + db_tag = self.get_tag_by_name(tag.name) + if db_tag is not None: + raise ValueError("Tried to create new tag with a name that is already taken!") + if tag.category_id is None: + raise Exception("Tried to create tag with no category!") - def delete_tag(self, tag_id: int): - raise NotImplementedError + if tag.implications is None: + tag.implications = [] + if tag.aliases is None: + tag.aliases = [] + + db_tag = DBTag(name=tag.name, description=tag.description, category_id=tag.category_id, + implications=tag.implications, alias=tag.aliases) + + db = self.__get_db() + db.add(db_tag) + db.commit() + + # TODO check that alias is bidirectional automatically + # e.g. Alias tag1 -> tag2 needs to also turn up as tag2 -> tag1 + + return db_tag + + def update_tag_by_model(self, tag: Tag): + db_tag = self.get_tag_by_id(tag.tag_id) + + db_categ = self.get_category_by_id(tag.category_id) + if db_categ is None: + raise ValueError("The given category id was not found!") + + db_tag.name = tag.name if tag.name is not None else db_tag.name + db_tag.description = tag.description if tag.description is not None else db_tag.description + db_tag.category_id = tag.category_id if tag.category_id is not None else db_tag.category_id + + db_tag.alias = tag.aliases if tag.aliases is not None else db_tag.alias + db_tag.implications = tag.implications if tag.implications is not None else db_tag.implications + + self.__get_db().commit() # TODO give feedback if alias, implication does not exist + + def delete_tag_by_id(self, tag_id: int): + db_tag = self.get_tag_by_id(tag_id) + if db_tag is None: + raise ValueError("Could not find the tag to delete!") + + db = self.__get_db() + db.delete(db_tag) + db.commit() + + def delete_tag_by_model(self, tag: Tag): + db_tag = self.get_tag_by_id(tag.tag_id) + if db_tag is None: + raise ValueError("Could not find the tag to delete!") + + db = self.__get_db() + db.delete(db_tag) + db.commit() def search_tag_by_name_fuzzy(self, search: str) -> list: # return a list of tags fitting the fuzzy name search - result = self.__get_db().query(DBTag).filter(DBTag.name.ilike("%{}%".format(search)))\ + result = self.__get_db().query(DBTag).filter(DBTag.name.ilike("%{}%".format(search))) \ .options(load_only("tag_id", "name")).all() return result diff --git a/database/models.py b/database/models.py index e94b9ad..339b5d2 100644 --- a/database/models.py +++ b/database/models.py @@ -53,17 +53,20 @@ class TagCategory(TagCategorynoID): orm_mode = True -class TagnoID(BaseModel): +class TagNoID(BaseModel): name: Optional[str] description: Optional[str] - category_id: Optional[TagCategory] + category_id: Optional[int] + + implications: Optional[List[int]] + aliases: Optional[List[int]] class Config: orm_mode = True -class Tag(TagnoID): - tag_ID: int +class Tag(TagNoID): + tag_id: int class Config: orm_mode = True diff --git a/main.py b/main.py index 80d3a5e..553d8c6 100644 --- a/main.py +++ b/main.py @@ -5,7 +5,7 @@ from typing import List, Tuple from database.database import Database from database.database import DBPresence, DBArt2Presence -from database.models import Art, ArtnoID, Presence, ArtistNoId, TagnoID, TagCategory, TagCategorynoID +from database.models import Art, ArtnoID, Presence, ArtistNoId, Tag, TagNoID, TagCategory, TagCategorynoID app = FastAPI() db = Database() @@ -225,16 +225,15 @@ async def topic(name: str = None, id: int = None, artist_id: int = None): 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})") - if id is None and name is None and category is None: - raise HTTPException(status_code=406) - 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 - raise HTTPException(status_code=404, detail="No matching tag found!") else: if id is not None: result = db.get_tag_by_id(id) @@ -250,13 +249,32 @@ async def tag(id: int = None, name: str = None, fuzzy: bool = False, category: i else: result.name = result.name.strip() return result - raise HTTPException(status_code=404, detail="Tag was not found!") + raise HTTPException(status_code=404, detail="No matching tag found!") @app.post("/artnet/metadata/tag") -async def tag(tag: TagnoID, id: int = None): +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).tag_id + return {"id": tag_id} + else: # update already existing tag + 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)) + + +@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): diff --git a/tests/createAPI/create_delete_tag.py b/tests/createAPI/create_delete_tag.py new file mode 100644 index 0000000..d140944 --- /dev/null +++ b/tests/createAPI/create_delete_tag.py @@ -0,0 +1,200 @@ +import requests +import json +from typing import List, Tuple + +from tests.createAPI.create_delete_tag_category import test_tag_category_entries, list_tag_categories, \ + create_tag_categories, delete_category_entries + +test_tag_entries = [{"name": "tag1", "description": "description1", "category_id": 0, "id": None}, + {"name": "tag2", "description": "description2", "category_id": 0, "id": None}, + {"name": "tag3", "description": "description3", "category_id": 0, "id": None}, + {"name": "tag4", "description": "description4", "category_id": 2, "id": None}] + + +def list_tags(url: str, port: int): + r = requests.get(f"http://{url}:{port}/artnet/metadata/tag") + if r.status_code != 200: + raise Exception("Failed querying for tags!") + tags = json.loads(r.text) + return tags + + +def create_tag_entries(url: str, port: int): + """ + Create many tags for testing purposes + :param url: + :param port: + :return: + """ + for i in range(len(test_tag_entries)): + r = create_tag(url, port, name=test_tag_entries[i]["name"], description=test_tag_entries[i]["description"], + category_id=test_tag_category_entries[test_tag_entries[i]["category_id"]]["id"]) + + if r.status_code != 200: + print(f"Create Tag Entry Test Nr.{i}: failed with {r.status_code} and reason {r.text}") + raise Exception("Create Tag Entry Test: FAILED") + else: + test_tag_entries[i]["id"] = json.loads(r.text)["id"] + + +def update_tag_entries(url: str, port: int): + """ + Update all tags in their fields and check the result + :param url: + :param port: + :return: + """ + for i in range(len(test_tag_entries) - 2): + new_name = test_tag_entries[i]["name"] + "_updated" + new_desc = test_tag_entries[i]["description"] + "_updated" + new_categ = test_tag_category_entries[1]["id"] + + r = update_tag(url, port, tag_id=test_tag_entries[i]["id"], name=new_name, description=new_desc, + category_id=new_categ) + + if r.status_code != 200: + print(f"Updating Tag Entry test Nr.{i} failed with {r.status_code} and reason {r.text}") + raise Exception("Update Tag Entry Test: FAILED") + + new_tag = get_tag_by_ID(url, port, test_tag_entries[i]["id"]) + + if new_tag["name"] != new_name: + print(f"Tag name is not matching the update name! Current: {new_tag['name']} Expected: {new_name}") + raise Exception("Update Tag Entry Test: FAILED") + if new_tag["description"] != new_desc: + print(f"Tag description is not matching the update description! " + f"Current: {new_tag['description']} Expected: {new_desc}") + raise Exception("Update Tag Entry Test: FAILED") + if new_tag["category_id"] != new_categ: + print(f"Tag category is not matching the update category! " + f"Current: {new_tag['category_id']} Expected: {new_categ}") + raise Exception("Update Tag Entry Test: FAILED") + + +def delete_tag_entries(url: str, port: int): + """ + Delete all tag entries specified by test_tag_entries + :param url: + :param port: + :return: + """ + for i in range(len(test_tag_entries)): + r = delete_tag(url, port, test_tag_entries[i]["id"]) + + if r.status_code != 200: + print(f"Delete Tag Entry Test Nr.{i}: failed with {r.status_code} and reason {r.text}") + raise Exception("Delete Tag Entry Test: FAILED") + + +def get_tag_by_ID(url: str, port: int, tag_id: int): + """ + Fetch a tag specified by tag_id + :param url: + :param port: + :param tag_id: + :return: + """ + r = requests.get(f"http://{url}:{port}/artnet/metadata/tag?id={tag_id}") + + if r.status_code != 200: + raise Exception("Failed querying for tag!") + + return json.loads(r.text) + + +def create_tag(url: str, port: int, name: str, category_id: int, description: str = None): + """ + Create a tag + :param url: + :param port: + :param name: + :param category_id: + :param description: + :return: + """ + r = requests.post(f"http://{url}:{port}/artnet/metadata/tag", + json={"name": name, "category_id": category_id, "description": description}) + + return r + + +def update_tag(url: str, port: int, tag_id: int, name: str = None, description: str = None, category_id: int = None, + implications: List[int] = None, alias: List[int] = None): + """ + Update a tag with the specified fields. If avoided they are not updated. + :param url: + :param port: + :param tag_id: + :param name: + :param description: + :param category_id: + :param implications: + :param alias: + :return: + """ + r = requests.post(f"http://{url}:{port}/artnet/metadata/tag?id={tag_id}", + json={"name": name, "description": description, "category_id": category_id, + "implications": implications, "alias": alias}) + return r + + +def delete_tag(url: str, port: int, tag_id: int): + """ + Delete a tag specified by its id + :param url: + :param port: + :param tag_id: + :return: + """ + r = requests.delete(f"http://{url}:{port}/artnet/metadata/tag?id={tag_id}") + + return r + + +def run_tag_test(url: str, port: int): + print() + print("----------------") + print(f"Starting tag test with ({len(list_tags(url, port))}) tags and " + f"({len(list_tag_categories(url, port))}) tag categories!") + print(f"Creating {len(test_tag_category_entries)} tag category entries for the tests ...") + create_tag_categories(url, port) + print(f"Creating {len(test_tag_entries)} tag entries ...") + create_tag_entries(url, port) + create_tag_result = False if not len(list_tags(url, port)) == len(test_tag_entries) else True + print(f"Found {len(list_tags(url, port))} tag entries!") + print() + + # Update test + print("Trying to update the tag fields ...") + update_tag_entries(url, port) + update_tag_result = True + print("Finished updating the tags!") + + print() + print(f"Found {len(list_tags(url, port))} tag entries!") + print("Deleting the tag entries ...") + delete_tag_entries(url, port) # would throw an exception if an error occurred + delete_tag_result = True + print(f"Found {len(list_tags(url, port))} tag entries!") + + # Clean up + print("Cleaning tag categories up ...") + delete_category_entries(url, port) + print(f"Tag test complete with {len(list_tags(url, port))} tags and " + f"{len(list_tag_categories(url, port))} categories!") + + return create_tag_result, update_tag_result, delete_tag_result # create, update, delete + + +if __name__ == "__main__": + url, port = "127.0.0.1", 8000 + l = len(list_tag_categories(url, port)) + if l > 0: + print(f"Deleting leftover category entries ({l}) ...") + delete_category_entries(url, port) + l = len(list_tags(url, port)) + if l > 0: + print(f"Deleting leftover tag entries ({l}) ...") + delete_tag_entries(url, port) + + run_tag_test(url, port) diff --git a/tests/run_tests.py b/tests/run_tests.py index 039e4d3..859b373 100644 --- a/tests/run_tests.py +++ b/tests/run_tests.py @@ -2,6 +2,7 @@ from tests.createAPI.create_delete_art import run_art_test, list_art, delete_art from tests.createAPI.create_delete_presence import run_presence_test, list_presences from tests.createAPI.create_delete_artist import delete_artist_entries, list_artists, run_artist_test from tests.createAPI.create_delete_tag_category import delete_category_entries, list_tag_categories, run_tag_category_test +from tests.createAPI.create_delete_tag import create_tag_entries, list_tags, run_tag_test, delete_tag_entries from tests.relationAPI.relate_art import run_art_presence_relation_test @@ -22,12 +23,16 @@ def run_tests(url: str, port: int): if len(list_tag_categories(url, port)) > 0: print("Deleting leftover tag categories ...") delete_category_entries(url, port) + if len(list_tags(url, port)) > 0: + print("Deleting leftover tags ...") + delete_tag_entries(url, port) create_artist_result, delete_artist_result = run_artist_test(url, port) create_presence_result, delete_presence_result, cascade_delete_presence_result = run_presence_test(url, port) create_art_result, update_art_result, delete_art_result = run_art_test(url, port) create_art2presence_result, delete_art2presence_result = run_art_presence_relation_test(url, port) create_category_result, update_category_result, delete_category_result = run_tag_category_test(url, port) + create_tag_result, update_tag_result, delete_tag_result = run_tag_test(url, port) print() print("-------- Test Results ---------") @@ -40,6 +45,8 @@ def run_tests(url: str, port: int): print(f"Art: {len(r)}, {r}") r = list_tag_categories(url, port) print(f"Tag Categories: {len(r)}, {r}") + r = list_tags(url, port) + print(f"Tags: {len(r)}, {r}") print() print("Functions:") print(f"Artists: \t\t\t\tCreate: {create_artist_result} \tUpdate: {'N/A'} " @@ -52,7 +59,8 @@ def run_tests(url: str, port: int): f"\tDelete: {delete_art2presence_result} \t(Direct)") print(f"Tag Category: \t\t\tCreate: {create_category_result} \tUpdate: {update_category_result} " f"\tDelete: {delete_category_result} \t(Direct)") - print(f"Tag: \t\t\t\t\tN/A") + print(f"Tag: \t\t\t\t\tCreate: {create_tag_result} \tUpdate: {update_tag_result} " + f"\tDelete: {delete_tag_result} \t(Direct)") print(f"(Artist) Topic: \t\tN/A") print(f"Art Collection: \t\tN/A") print(f"Art2Art Collection: \tN/A")