Implemented Art Collection calls and tests

Implemented collection calls and test script. Does not include actual relations between collections and art.

database/database.py

@@ -6,7 +6,8 @@ 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, TagNoID, Tag, Topic, TopicNoId
+from database.models import Art, ArtnoID, Artist, Presence, TagCategory, TagCategorynoID, TagNoID, Tag, \
+    Topic, TopicNoId, Collection, CollectionNoID
 
 # TODO read sensitive data from environment file or similar
 SQLALCHEMY_DATABASE_URL = "postgresql://artnet_editor:G606Rm9sFEXe6wfTLxVu@[::1]/test_artnet"
@@ -575,3 +576,44 @@ class Database:
             self.__get_db().commit()
         except psycopg2.IntegrityError as e:
             print(e)
+
+    # Collection
+    def get_collection_list(self) -> List[DBCollection]:
+        result = self.__get_db().query(DBCollection).all()  # TODO fix StackOverflow
+        return result
+
+    def get_collection_by_id(self, id: int):
+        result = self.__get_db().query(DBCollection).where(DBCollection.id == id).first()
+        return result
+
+    def get_collections_by_name(self, name: str):
+        result = self.__get_db().query(DBCollection).where(DBCollection.name == name)
+        return result
+
+    def create_collection_by_model(self, coll: CollectionNoID):
+        db_collection = DBCollection(name=coll.name, description=coll.description)
+
+        self.__get_db().add(db_collection)
+        self.__get_db().commit()
+
+        return db_collection
+
+    def update_collection_by_model(self, coll: Collection):
+        db_collection = self.get_collection_by_id(coll.id)
+        if db_collection is None:
+            raise ValueError("Could not find a collection with this ID!")
+
+        db_collection.name = coll.name
+        db_collection.description = coll.description
+
+        self.__get_db().commit()
+
+    def delete_collection_by_id(self, id: int):
+        db_collection = self.get_collection_by_id(id)
+        if db_collection is None:
+            raise ValueError("Could not find a collection with this ID!")
+
+        self.__get_db().delete(db_collection)
+        self.__get_db().commit()
+
+

database/models.py

@@ -91,3 +91,17 @@ class Art(ArtnoID):
     class Config:
         orm_mode = True
 
+
+class CollectionNoID(BaseModel):
+    name: str
+    description: Optional[str]
+
+    class Config:
+        orm_mode = True
+
+
+class Collection(CollectionNoID):
+    id: int
+
+    class Config:
+        orm_mode = True

main.py

@@ -5,7 +5,8 @@ from typing import List, Tuple
 
 from database.database import Database
 from database.database import DBPresence, DBArt2Presence
-from database.models import Art, ArtnoID, Presence, ArtistNoId, Tag, TagNoID, TagCategory, TagCategorynoID, TopicNoId, Topic
+from database.models import Art, ArtnoID, Presence, ArtistNoId, Tag, TagNoID, TagCategory, TagCategorynoID, TopicNoId, \
+    Topic, CollectionNoID, Collection
 
 app = FastAPI()
 db = Database()
@@ -290,6 +291,8 @@ async def tag(tag: TagNoID, id: int = None):
         except ValueError as e:
             raise HTTPException(status_code=404, detail=str(e))
 
+        return True
+
 
 @app.delete("/artnet/metadata/tag")
 async def tag(id: int):
@@ -346,5 +349,52 @@ def category(id: int):
         raise HTTPException(status_code=404, detail=str(e))
 
 
+@app.get("/artnet/metadata/collection")
+def collection(id: int = None, name: str = None):
+    print(f"Received GET on /artnet/metadata/collection (id={id}, name={name})")
+    result = None
+    if id is None and name 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 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
+
+
+@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))
+
+
 if __name__ == "__main__":
     uvicorn.run(app, host="127.0.0.1", port=8000)

tests/createAPI/create_delete_collection.py

@@ -0,0 +1,122 @@
+import requests
+import json
+
+test_collection_entries = [{"name": "collection_name1", "description": "description1", "id": None},
+                           {"name": "collection_name2", "description": "description2", "id": None}]
+
+
+def list_collections(url: str, port: int):
+    r = requests.get(f"http://{url}:{port}/artnet/metadata/collection")
+    if r.status_code != 200:
+        print(f"Querying collections failed! Status: {r.status_code} Reason: {r.text}")
+        raise Exception("Failed to query for collections!")
+
+    collections = json.loads(r.text)
+    return collections
+
+
+def create_collection_entries(url: str, port: int):
+    for i in range(len(test_collection_entries)):
+        r = create_collection(url, port, name=test_collection_entries[i]["name"],
+                              description=test_collection_entries[i]["description"])
+
+        if r.status_code != 200:
+            print(f"Create Collection Entry Test Nr.{i}: failed with {r.status_code} and reason {r.text}")
+            raise Exception("Create Collection Entry Test: FAILED")
+        else:
+            test_collection_entries[i]["id"] = json.loads(r.text)["id"]
+
+
+def update_collection_entries(url: str, port: int):
+    for i in range(len(test_collection_entries)):
+        new_name = test_collection_entries[i]["name"] + "_updated"
+        new_desc = test_collection_entries[i]["description"] + "_updated"
+
+        r = update_collection(url, port, test_collection_entries[i]["id"], new_name, new_desc)
+        if r.status_code != 200:
+            print(f"Updating Collection Entry Test Nr.{i}: failed with {r.status_code} and reason {r.text}")
+            raise Exception("Update Collection Entry Test: FAILED")
+
+        new_coll = get_collection_by_id(url, port, test_collection_entries[i]["id"])
+
+        if new_coll["name"] != new_name:
+            print(f"Collection name is not matching the update collection! "
+                  f"Current: {new_coll['name']} Expected: {new_name}")
+            raise Exception("Update Collection Entry Test: FAILED")
+        if new_coll["description"] != new_desc:
+            print(f"Collection description is not matching the update collection! "
+                  f"Current: {new_coll['description']} Expected: {new_desc}")
+            raise Exception("Update Collection Entry Test: FAILED")
+
+
+def delete_collection_entries(url: str, port: int):
+    for i in range(len(test_collection_entries)):
+        r = delete_collection(url, port, id=test_collection_entries[i]["id"])
+        if r.status_code != 200:
+            print(f"Delete Collection Entry Test Nr.{i}: failed with {r.status_code} and reason {r.text}")
+            raise Exception("Delete Collection Entry Test: FAILED")
+
+
+def delete_all_collections(url: str, port: int):
+    collections = list_collections(url, port)
+    for collection in collections:
+        r = delete_collection(url, port, collection["id"])
+        if r.status_code != 200:
+            print(f"Failed deleting collections! Status: {r.status_code} Reason: {r.text}")
+            raise Exception("Failed deleting collection!")
+
+
+def get_collection_by_id(url: str, port: int, id: str):
+    r = requests.get(f"http://{url}:{port}/artnet/metadata/collection?id={id}")
+    if r.status_code != 200:
+        raise Exception("Failed querying for collection!")
+
+    return json.loads(r.text)
+
+
+def get_collection_by_name(url: str, port: int, name: str):
+    r = requests.get(f"http://{url}:{port}/artnet/metadata/collection?name={name}")
+    return r
+
+
+def create_collection(url: str, port: int, name: str, description: str):
+    r = requests.post(f"http://{url}:{port}/artnet/metadata/collection",
+                      json={"name": name, "description": description})
+    return r
+
+
+def update_collection(url: str, port: int, id: str, name: str, description: str):
+    r = requests.post(f"http://{url}:{port}/artnet/metadata/collection?id={id}",
+                      json={"name": name, "description": description})
+    return r
+
+
+def delete_collection(url: str, port: int, id: str):
+    r = requests.delete(f"http://{url}:{port}/artnet/metadata/collection?id={id}")
+    return r
+
+
+def run_collection_tests(url: str, port: int):
+    print()
+    print("----------------")
+    l = len(list_collections(url, port))
+    print(f"Starting collection test with ({l}) collections!")
+    if l > 0:
+        print("Deleting leftover collections ...")
+        delete_all_collections(url, port)
+
+    print(f"Creating {len(list_collections(url, port))} collections as a test ...")
+    create_collection_entries(url, port)
+    create_collection_result = False if not len(list_collections(url, port)) else True
+    print(f"Found {len(list_collections(url, port))} collection entries!")
+
+    print(f"Updating collection entries with new data ...")
+    update_collection_entries(url, port)
+    update_collection_result = True
+
+    print(f"Deleting the collections again ...")
+    delete_collection_entries(url, port)
+    delete_collection_result = False if not len(list_collections(url, port)) == 0 else True
+    print(f"Found {len(list_collections(url, port))} collection entries!")
+
+    return create_collection_result, update_collection_result, delete_collection_result

tests/createAPI/create_delete_topic.py

@@ -1,6 +1,5 @@
 import requests
 import json
-from typing import List, Tuple
 
 from tests.createAPI.create_delete_artist import create_artist_entries, test_artist_entries, delete_artist_entries, \
     list_artists

tests/run_tests.py

@@ -1,9 +1,11 @@
 from tests.createAPI.create_delete_art import run_art_test, list_art, delete_art_entries
 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_category import delete_category_entries, list_tag_categories, \
+    run_tag_category_test
 from tests.createAPI.create_delete_tag import list_tags, run_tag_test, delete_tag_entries
 from tests.createAPI.create_delete_topic import list_topics, run_topic_tests, delete_all_topics
+from tests.createAPI.create_delete_collection import run_collection_tests
 
 from tests.relationAPI.relate_art import run_art_presence_relation_test
 
@@ -35,6 +37,7 @@ def run_tests(url: str, port: int):
     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)
     create_topic_result, update_topic_result, delete_topic_result = run_topic_tests(url, port)
+    create_collection_result, update_collection_result, delete_collection_result = run_collection_tests(url, port)
 
     print()
     print("-------- Test Results ---------")
@@ -68,7 +71,8 @@ def run_tests(url: str, port: int):
           f"\tDelete: {delete_tag_result} \t(Direct)")
     print(f"(Artist) Topic: \t\tCreate: {create_topic_result} \tUpdate: {update_topic_result} "
           f"\tDelete: {delete_topic_result} \t(Direct)")
-    print(f"Art Collection: \t\tN/A")
+    print(f"Art Collection: \t\tCreate: {create_collection_result} \tUpdate: {update_collection_result} "
+          f"\tDelete: {delete_collection_result} \t(Direct)")
     print(f"Art2Art Collection: \tN/A")
     print(f"Artist2Topic: \t\t\tN/A")
     print(f"Art2Tag: \t\t\t\tN/A")