ArtNet-Webservice/database/database.py

from typing import List

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

# TODO read sensitive data from environment file or similar
SQLALCHEMY_DATABASE_URL = "postgresql://artnet_editor:G606Rm9sFEXe6wfTLxVu@127.0.0.1/test_artnet"
engine = sqlalchemy.create_engine(SQLALCHEMY_DATABASE_URL, echo=True)
SessionLocal = sqlalchemy.orm.sessionmaker(autocommit=False, autoflush=False, bind=engine)


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


class DBPresence(Base):
    __tablename__ = "presence"

    name = Column(String(20), primary_key=True)
    domain = Column(String(20), primary_key=True)
    link = Column(String, nullable=True)
    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",
                        cascade="delete, all")


class DBArt(Base):
    __tablename__ = 'art'

    id = Column(Integer, primary_key=True, index=True)
    md5_hash = Column(String(32), nullable=False, unique=True)
    path = Column(String, nullable=False, unique=True)
    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")


class DBArtPresence(Base):
    __tablename__ = "art_author"

    presence_name = Column(String(20), primary_key=True)
    presence_domain = Column(String(20), primary_key=True)
    art_id = Column(Integer, ForeignKey(DBArt.id, ondelete="CASCADE"), primary_key=True)

    __table_args__ = (
        ForeignKeyConstraint(
            ("presence_name", "presence_domain"),
            ["presence.name", "presence.domain"]
        ),
    )


class DBArtist(Base):
    __tablename__ = "artist"

    id = Column(Integer, primary_key=True)
    description = Column(String, nullable=False)

    topics = relationship("DBTopic", secondary="artist_topic", back_populates="artists", viewonly=True)
    presences = relationship("DBPresence", back_populates="artist", cascade="all, delete")


class DBTopic(Base):  # as of now unused
    __tablename__ = "topic"

    id = Column(Integer, primary_key=True)
    name = Column(String(20), unique=True, nullable=False)
    description = Column(String)

    artists = relationship("DBArtist", secondary="artist_topic", back_populates="topics", viewonly=True)


class DBTagCategory(Base):
    __tablename__ = "tag_category"

    category_id = Column(Integer, primary_key=True)
    name = Column(String(20), nullable=False)


class DBTag(Base):
    __tablename__ = "tag"

    tag_id = Column(Integer, primary_key=True)
    name = Column(String(50), unique=True)
    description = Column(String)
    category_id = Column(Integer, ForeignKey('tag_category.category_id'))

    category = relationship("DBTagCategory", backref='tags', foreign_keys=[category_id])
    # TODO check if cascade is required
    art = relationship("DBArt", secondary=art_tag_table, back_populates="tags")
    # TODO check if cascade is required


Base.metadata.create_all(bind=engine)


class Database:

    __db: sqlalchemy.orm.Session = None

    def __get_db(self) -> sqlalchemy.orm.Session:
        Database.__db = SessionLocal() if Database.__db is None else Database.__db
        try:
            return Database.__db
        finally:
            ##Database.__db.close()
            pass

    def __del__(self):
        Database.__db.close()
        Database.__db = None

# Art

    def get_art_list(self):
        return self.__get_db().query(DBArt).all()  # TODO FIX StackOverflow

    def get_art_by_id(self, art_id: int):
        return self.__get_db().query(DBArt).where(DBArt.id == art_id).first()

    def get_art_by_hash(self, md5_hash: str):
        return self.__get_db().query(DBArt).filter(func.lower(DBArt.md5_hash) == md5_hash.lower()).first()

    def create_art_by_model(self, art: ArtnoID):
        if not (isinstance(art.path, str) and len(art.path) > 0):
            raise ValueError("New Art must contain a path!")

        db_art = DBArt(title=art.title, md5_hash=art.hash, path=art.path, link=art.link)

        if self.get_art_by_hash(art.hash) is not None:
            raise ValueError("New Art must not contain already known hashes! Is this really new?")

        db = self.__get_db()
        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.add(db_art_presence)
            db.commit()

        return db_art.id

    @DeprecationWarning
    def update_art_by_id(self, art_id: int, title: str, path: str, tags: list, link: str,
                         md5_hash: str, authors: list = None):
        raise NotImplementedError

    @DeprecationWarning
    def update_art_by_hash(self, md5_hash: str, title: str, path: str, tags: list, link: str, authors: list = None):
        raise NotImplementedError

    def update_art_by_model(self, art: Art):
        db_art: DBArt = self.get_art_by_id(art.id)

        db_art.title = art.title if art.title is not None else db_art.title
        db_art.link = art.link if art.link is not None else db_art.link
        db_art.md5_hash = art.hash if art.hash is not None else db_art.md5_hash
        db_art.path = art.path if art.path is not None else db_art.path

        self.__get_db().commit()

    def delete_art_by_id(self, art_id: int):
        db_art: DBArt = self.get_art_by_id(art_id)
        self.__get_db().delete(db_art)
        self.__get_db().commit()

    @DeprecationWarning
    def delete_art_by_hash(self, md5_hash: str):
        raise NotImplementedError

    # 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()
        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()
        return result

    def update_art_presences_by_id(self, art_id: int, presences: list):  # presences = [("name", "domain"),(...)]
        raise NotImplementedError

    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

    def delete_art_presences_by_hash(self, md5_hash: str, presence_name: str, presence_domain: str):
        raise NotImplementedError

    # Presence

    def get_presence_list(self):
        return self.__get_db().query(DBPresence).all()  # TODO fix StackOverflow

    def get_presence(self, name: str, domain: str) -> DBPresence:
        result = self.__get_db().query(DBPresence)\
            .filter(func.lower(DBPresence.name) == name.lower() and
                    func.lower(DBPresence.domain) == domain.lower()).first()
        return result

    def create_presence(self, name: str, domain: str, artist_id: int, link: str):
        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!")

        db_presence = DBPresence(name=name, domain=domain, artist_id=artist_id, link=link)

        db = self.__get_db()
        db.add(db_presence)
        db.commit()

        return db_presence.name, db_presence.domain

    def update_presence(self, name: str, domain: str, artist_id: int, link: str):
        db_presence = self.get_presence(name=name, domain=domain)

        db_presence.link = link
        db_presence.artist_id = artist_id

        self.__get_db().commit()

    def delete_presence(self, name: str, domain: str):
        db_presence = self.get_presence(name=name, domain=domain)
        self.__get_db().delete(db_presence)
        self.__get_db().commit()

    # Artist -> Presence

    def get_artist_presences(self, artist_id: int):
        result = self.__get_db().query(DBPresence).filter(DBPresence.artist_id == artist_id).all()
        return result

    # Artist

    def get_artist_list(self):
        return self.__get_db().query(DBArtist).all()  # TODO fix StackOverflow

    def get_artist(self, artist_id: int) -> DBArtist:
        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)

        db = self.__get_db()
        db.add(db_artist)
        db.commit()

        return db_artist

    def search_artist(self, name: str) -> Artist:
        result = self.__get_db().query(DBArtist).where(DBArtist.description == name).all()
        return result

    def update_artist(self, artist_id: int, name: str = None, topics: List[int] = None):
        db_artist = self.get_artist(artist_id)

        if name is not None:
            db_artist.description = name
        if topics is not None:
            db_artist.topics = topics

        self.__get_db().commit()

    def delete_artist(self, artist_id: int):
        db_artist = self.get_artist(artist_id)

        self.__get_db().delete(db_artist)
        self.__get_db().commit()  # TODO FIX sqlalchemy.exc.IntegrityError: (psycopg2.errors.ForeignKeyViolation)
                                    # TODO update or delete on table "artist" violates foreign key constraint "presence_artist_id_fkey" on table "presence"
                                    # TODO DETAIL:  Key (id)=(83) is still referenced from table "presence".

    # Topic

    def get_topic_by_id(self, id: int):
        result = self.__get_db().query(DBTopic).filter(DBTopic.id == id).first()
        return result

    def get_topic_by_name(self, name: str):
        result = self.__get_db().query(DBTopic).filter(func.lower(DBTopic.name) == name.lower()).first()
        return result

    def update_topic(self, name: str, description: str):
        raise NotImplementedError

    def delete_topic(self, name: str):
        raise NotImplementedError

    # Artist -> Topic

    def get_artist_topics(self, artist_id: int):
        result = self.__get_db().query(DBTopic).filter(DBTopic.artists.any(id=artist_id)).all()
        return result

    def update_artist_topics(self, artist_id: int, topic_ids: list):
        raise NotImplementedError

    def delete_artist_topics(self, artist_id: int, topic_ids: list):
        raise NotImplementedError

    # Topic -> Artist

    def get_topic_artists(self, topic_id: int):
        result = self.__get_db().query(DBArtist).filter(DBArtist.topics.any(id=topic_id)).all()
        return result

    def update_topic_artists(self, topic_id: int, artists: list):
        raise NotImplementedError

    def delete_topic_artists(self, topic_id: int):  # deletes only the connections, not the artists
        raise NotImplementedError

    # Tag

    def get_tag_by_id(self, tag_id: int):
        result = self.__get_db().query(DBTag).where(tag_id == DBTag.tag_id).first()
        return result

    def get_tag_by_name(self, name: str):
        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 delete_tag(self, tag_id: int):
        raise NotImplementedError

    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)))\
            .options(load_only("tag_id", "name")).all()
        return result

    # Tag -> Art

    def get_tag_art(self, tag_id: int):
        result = self.__get_db().query(DBArt).filter(DBArt.tags.any(tag_id=tag_id)).all()
        return result

    def update_tag_art(self, tag_id: int):  # is this useful?
        raise NotImplementedError

    def delete_tag_art(self):  # deletes only the connections, not the art
        raise NotImplementedError

    # Category -> Tag

    def get_category_tags(self, category_id: int):
        result = self.__get_db().query(DBTag).where(DBTag.category_id == category_id).all()
        return result

    def update_category_tags(self, category_id: int, tags: list):
        raise NotImplementedError

    def delete_category_tags(self, category_id: int):
        raise NotImplementedError

    # Category

    def get_category_by_id(self, category_id: int):
        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()
        return result

    def update_category(self, category_id: int):
        raise NotImplementedError

    def delete_category(self, category_id: int):
        raise NotImplementedError