Added Tag models and basic tests

Added SQLAlchemy models for tag alias and implication relations.
Added test scripts to test basic tag functionalities. This does not include implication and alias which will have their own test scripts.

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
 

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

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):

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)

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")