You cannot select more than 25 topics Topics must start with a letter or number, can include dashes ('-') and can be up to 35 characters long.

396 lines
14 KiB
Python

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