diff --git a/database/database.py b/database/database.py index 6671d16..8d1eda4 100644 --- a/database/database.py +++ b/database/database.py @@ -1,13 +1,14 @@ from typing import List +import psycopg2 from sqlalchemy import Column, Boolean, Float, String, Integer, ForeignKey, Table, func, ForeignKeyConstraint import sqlalchemy from sqlalchemy.orm import declarative_base, relationship, load_only -from database.models import Art, ArtnoID, Artist +from database.models import Art, ArtnoID, Artist, Presence, TagCategory, TagCategorynoID # TODO read sensitive data from environment file or similar -SQLALCHEMY_DATABASE_URL = "postgresql://artnet_editor:G606Rm9sFEXe6wfTLxVu@127.0.0.1/test_artnet" +SQLALCHEMY_DATABASE_URL = "postgresql://artnet_editor:G606Rm9sFEXe6wfTLxVu@[::1]/test_artnet" engine = sqlalchemy.create_engine(SQLALCHEMY_DATABASE_URL, echo=True) SessionLocal = sqlalchemy.orm.sessionmaker(autocommit=False, autoflush=False, bind=engine) @@ -15,12 +16,18 @@ SessionLocal = sqlalchemy.orm.sessionmaker(autocommit=False, autoflush=False, bi Base = sqlalchemy.orm.declarative_base() -art_tag_table = Table('art_tag', Base.metadata, - Column('art_id', Integer, ForeignKey('art.id')), - Column('tag_id', Integer, ForeignKey('tag.tag_id'))) -artist_topic_table = Table('artist_topic', Base.metadata, - Column('artist_id', Integer, ForeignKey('artist.id')), - Column('topic_id', Integer, ForeignKey('topic.id'))) +art_to_tag_table = Table('art_to_tag', Base.metadata, + Column('art_id', Integer, ForeignKey('art.id')), + Column('tag_id', Integer, ForeignKey('tag.tag_id'))) + +art_to_art_collection = Table('art_to_art_collection', Base.metadata, + Column('collection_id', Integer, ForeignKey('art_collection.id')), + Column('art_id', Integer, ForeignKey('art.id')), + Column('ranking', String)) + +artist_to_topic_table = Table('artist_to_topic', Base.metadata, + Column('artist_id', Integer, ForeignKey('artist.id')), + Column('topic_id', Integer, ForeignKey('topic.id'))) class DBPresence(Base): @@ -32,7 +39,7 @@ class DBPresence(Base): artist_id = Column(Integer, ForeignKey('artist.id', ondelete="CASCADE"), nullable=False) artist = relationship("DBArtist", back_populates="presences") - arts = relationship("DBArt", secondary="art_author", back_populates="presences", + arts = relationship("DBArt", secondary="art_to_presence", back_populates="presences", cascade="delete, all") @@ -45,13 +52,13 @@ class DBArt(Base): title = Column(String, nullable=True) link = Column(String, nullable=True) - tags = relationship("DBTag", secondary=art_tag_table, back_populates="art", viewonly=True) - presences = relationship("DBPresence", secondary="art_author", back_populates="arts", - cascade="all, delete") + tags = relationship("DBTag", secondary=art_to_tag_table, back_populates="art", viewonly=True) + collections = relationship("DBCollection", secondary=art_to_art_collection, back_populates="art") + presences = relationship("DBPresence", secondary="art_to_presence", back_populates="arts") -class DBArtPresence(Base): - __tablename__ = "art_author" +class DBArt2Presence(Base): + __tablename__ = "art_to_presence" presence_name = Column(String(20), primary_key=True) presence_domain = Column(String(20), primary_key=True) @@ -69,12 +76,22 @@ class DBArtist(Base): __tablename__ = "artist" id = Column(Integer, primary_key=True) - description = Column(String, nullable=False) + name = Column(String, nullable=False) - topics = relationship("DBTopic", secondary="artist_topic", back_populates="artists", viewonly=True) + topics = relationship("DBTopic", secondary="artist_to_topic", back_populates="artists", viewonly=True) presences = relationship("DBPresence", back_populates="artist", cascade="all, delete") +class DBCollection(Base): + __tablename__ = "art_collection" + + id = Column(Integer, primary_key=True) + name = Column(String, unique=True, nullable=False) + description = Column(String) + + art = relationship("DBArt", secondary=art_to_art_collection, back_populates="collections") + + class DBTopic(Base): # as of now unused __tablename__ = "topic" @@ -82,7 +99,7 @@ class DBTopic(Base): # as of now unused name = Column(String(20), unique=True, nullable=False) description = Column(String) - artists = relationship("DBArtist", secondary="artist_topic", back_populates="topics", viewonly=True) + artists = relationship("DBArtist", secondary="artist_to_topic", back_populates="topics", viewonly=True) class DBTagCategory(Base): @@ -91,6 +108,8 @@ class DBTagCategory(Base): category_id = Column(Integer, primary_key=True) name = Column(String(20), nullable=False) + tags = relationship("DBTag", back_populates="category", cascade="all, delete") + class DBTag(Base): __tablename__ = "tag" @@ -100,9 +119,9 @@ class DBTag(Base): description = Column(String) category_id = Column(Integer, ForeignKey('tag_category.category_id')) - category = relationship("DBTagCategory", backref='tags', foreign_keys=[category_id]) + category = relationship("DBTagCategory", back_populates='tags', foreign_keys=[category_id]) # TODO check if cascade is required - art = relationship("DBArt", secondary=art_tag_table, back_populates="tags") + art = relationship("DBArt", secondary=art_to_tag_table, back_populates="tags") # TODO check if cascade is required @@ -127,13 +146,13 @@ class Database: # Art - def get_art_list(self): + def get_art_list(self) -> List[DBArt]: return self.__get_db().query(DBArt).all() # TODO FIX StackOverflow - def get_art_by_id(self, art_id: int): + def get_art_by_id(self, art_id: int) -> DBArt: return self.__get_db().query(DBArt).where(DBArt.id == art_id).first() - def get_art_by_hash(self, md5_hash: str): + def get_art_by_hash(self, md5_hash: str) -> DBArt: return self.__get_db().query(DBArt).filter(func.lower(DBArt.md5_hash) == md5_hash.lower()).first() def create_art_by_model(self, art: ArtnoID): @@ -149,15 +168,14 @@ class Database: db.add(db_art) db.commit() - # TODO implement saving art <-> presence relationship as given by art.presences if art.presences is not None: for presence in art.presences: - db_art_presence = DBArtPresence(presence_name=presence.name, presence_domain=presence.domain, - art_id=db_art.id) + db_art_presence = DBArt2Presence(presence_name=presence.name, presence_domain=presence.domain, + art_id=db_art.id) db.add(db_art_presence) db.commit() - return db_art.id + return db_art @DeprecationWarning def update_art_by_id(self, art_id: int, title: str, path: str, tags: list, link: str, @@ -189,30 +207,59 @@ class Database: # Art -> Presences - def get_art_presences_by_id(self, art_id: int): - result = self.__get_db().query(DBArtPresence).filter(DBArtPresence.art_id == art_id).all() + def get_art_presences_by_id(self, art_id: int) -> List[DBArt2Presence]: + result = self.__get_db().query(DBArt2Presence).filter(DBArt2Presence.art_id == art_id).all() return result def get_art_presences_by_hash(self, md5_hash: str): - result = self.__get_db().query(DBArtPresence).join(DBArt).filter(DBArt.md5_hash == md5_hash).all() + result = self.__get_db().query(DBArt2Presence).join(DBArt).filter(DBArt.md5_hash == md5_hash).all() return result - def update_art_presences_by_id(self, art_id: int, presences: list): # presences = [("name", "domain"),(...)] - raise NotImplementedError + @DeprecationWarning # is this actually needed? superceeded by updating the art.presence field on art + def update_art_presences_by_id(self, art_id: int, presences: List[Presence]): # presences = [("name", "domain"),(...)] + """ + Creates an art-presence relation for every presence listed + :param art_id: + :param presences: + :return: + """ + db = self.__get_db() + + for presence in presences: + art2presence = DBArt2Presence() + art2presence.art_id = art_id + art2presence.presence_name = presence.name + art2presence.presence_domain = presence.domain + + db.add(art2presence) + + db.commit() def update_art_presences_by_hash(self, md5_hash: int, presences: list): # presences = [("name", "domain"),(...)] raise NotImplementedError def delete_art_presences_by_id(self, art_id: int, presence_name: str, presence_domain: str): - raise NotImplementedError + art2presences = self.get_art_presences_by_id(art_id) + + art2presence = None + for rel in art2presences: + if rel.presence_name.strip() == presence_name.strip() and \ + rel.presence_domain.strip() == presence_domain.strip(): + art2presence = rel + + if art2presence is None: + raise ValueError("Unknown art-presence relation") + + self.__get_db().delete(art2presence) + self.__get_db().commit() def delete_art_presences_by_hash(self, md5_hash: str, presence_name: str, presence_domain: str): raise NotImplementedError # Presence - def get_presence_list(self): + def get_presence_list(self) -> List[DBPresence]: return self.__get_db().query(DBPresence).all() # TODO fix StackOverflow def get_presence(self, name: str, domain: str) -> DBPresence: @@ -221,7 +268,7 @@ class Database: func.lower(DBPresence.domain) == domain.lower()).first() return result - def create_presence(self, name: str, domain: str, artist_id: int, link: str): + def create_presence(self, name: str, domain: str, artist_id: int, link: str) -> DBPresence: if not (len(name) > 0 and len(domain) > 0): print(f"Name: \"{name}\" Domain: \"{domain}\"") raise ValueError("New Presence must have some name and domain!") @@ -232,7 +279,7 @@ class Database: db.add(db_presence) db.commit() - return db_presence.name, db_presence.domain + return db_presence def update_presence(self, name: str, domain: str, artist_id: int, link: str): db_presence = self.get_presence(name=name, domain=domain) @@ -262,8 +309,8 @@ class Database: result = self.__get_db().query(DBArtist).where(DBArtist.id == artist_id).first() return result - def create_artist(self, name: str, topics: List[int]): - db_artist = DBArtist(description=name, topics=topics) + def create_artist(self, name: str, topics: List[int]) -> DBArtist: + db_artist = DBArtist(name=name, topics=topics) db = self.__get_db() db.add(db_artist) @@ -295,11 +342,11 @@ class Database: # Topic - def get_topic_by_id(self, id: int): + def get_topic_by_id(self, id: int) -> DBTopic: result = self.__get_db().query(DBTopic).filter(DBTopic.id == id).first() return result - def get_topic_by_name(self, name: str): + def get_topic_by_name(self, name: str) -> DBTopic: result = self.__get_db().query(DBTopic).filter(func.lower(DBTopic.name) == name.lower()).first() return result @@ -335,11 +382,11 @@ class Database: # Tag - def get_tag_by_id(self, tag_id: int): + 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 - def get_tag_by_name(self, name: str): + def get_tag_by_name(self, name: str) -> DBTag: result = self.__get_db().query(DBTag).filter(func.lower(DBTag.name) == name.lower()).first() return result @@ -379,17 +426,48 @@ class Database: raise NotImplementedError # Category + def get_category_list(self) -> List[DBTagCategory]: + result = self.__get_db().query(DBTagCategory).all() # TODO fix StackOverflow + return result - def get_category_by_id(self, category_id: int): + def get_category_by_id(self, category_id: int) -> DBTagCategory: result = self.__get_db().query(DBTagCategory).where(DBTagCategory.category_id == category_id).first() return result - def get_category_by_name(self, category: str): - result = self.__get_db().query(DBTagCategory).filter(func.lower(DBTagCategory.name) == category.lower()).first() + def get_category_by_name(self, name: str) -> DBTagCategory: + result = self.__get_db().query(DBTagCategory).filter(func.lower(DBTagCategory.name) == name.lower()).first() return result - def update_category(self, category_id: int): - raise NotImplementedError + def create_category_by_model(self, category: TagCategorynoID) -> DBTagCategory: + if not (isinstance(category.name, str) and len(category.name) > 0): + raise ValueError("New Category must bear a name!") + if self.get_category_by_name(category.name) is not None: + raise ValueError("New Tag Category must not contain an already used name! Is this a duplicate?") - def delete_category(self, category_id: int): - raise NotImplementedError + db_category = DBTagCategory(name=category.name) + + db = self.__get_db() + db.add(db_category) + db.commit() + + return db_category + + def update_category_by_model(self, category: TagCategory): + db_category: DBTagCategory = self.get_category_by_id(category_id=category.category_id) + if self.get_category_by_name(category.name) is not None: + raise ValueError("New Tag Category must not contain an already used name! " + f"Is this ({category.name}) a duplicate?") + + db_category.name = category.name + + self.__get_db().commit() + + def delete_category_by_id(self, category_id: int): + db_category: DBTagCategory = self.get_category_by_id(category_id) + if db_category is None: + raise ValueError(f"Tried to delete unknown category. ID ({category_id}) was not found!") + try: + self.__get_db().delete(db_category) + self.__get_db().commit() + except psycopg2.IntegrityError as e: + print(e)