ArtNet-Webservice/database/database.py

from sqlalchemy import Column, Boolean, Float, String, Integer, ForeignKey, Table, func, ForeignKeyConstraint
import sqlalchemy
from sqlalchemy.orm import declarative_base, relationship, load_only

# TODO read sensitive data from environment file or similar
SQLALCHEMY_DATABASE_URL = "postgresql://artnet_editor:G606Rm9sFEXe6wfTLxVu@127.0.0.1/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'), nullable=False, index=True)

    artist = relationship("DBArtist", backref='presences', foreign_keys=[artist_id])
    #artists = relationship('DBArtist', foreign_keys='DBArtist.id', back_populates="presences")
    arts = relationship("DBArt", secondary="art_author", viewonly=True)


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")
    presences = relationship("DBPresence", secondary="art_author", viewonly=True)


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), 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=True)

    topics = relationship("DBTopic", secondary="artist_topic", back_populates="artists")


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


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])
    art = relationship("DBArt", secondary=art_tag_table, back_populates="tags")


Base.metadata.create_all(bind=engine)


class Database:

    def __get_db(self) -> sqlalchemy.orm.Session:
        db = SessionLocal()
        try:
            return db
        finally:
            db.close()

# Art

    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 update_art_by_id(self, art_id: int, title: str, authors: list, path: str, tags: list, link: str,
                         md5_hash: str):
        raise NotImplementedError

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

    def delete_art_by_id(self, art_id: int):
        raise NotImplementedError

    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(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 update_presence(self, name: str, domain: str, artist_id: int, link: str):
        raise NotImplementedError

    def delete_presence(self, name: str, domain: str):
        raise NotImplementedError

    # 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(self, artist_id: int):
        result = self.__get_db().query(DBArtist).where(DBArtist.id == artist_id).first()
        return result

    def update_artist(self, artist_id: int, name: str):
        raise NotImplementedError

    def delete_artist(self, artist_id: int):
        raise NotImplementedError

    # 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