diff --git a/database/__init__.py b/database/__init__.py new file mode 100644 index 0000000..e69de29 diff --git a/database/database.py b/database/database.py new file mode 100644 index 0000000..652cc14 --- /dev/null +++ b/database/database.py @@ -0,0 +1,288 @@ +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 diff --git a/database/models.py b/database/models.py new file mode 100644 index 0000000..ff07135 --- /dev/null +++ b/database/models.py @@ -0,0 +1,13 @@ +from pydantic import BaseModel +from typing import List, Optional + + +class Art(BaseModel): + id: int + hash: str + path: str + title: Optional[str] = None + link: Optional[str] = None + + class Config: + orm_mode = True diff --git a/main.py b/main.py new file mode 100644 index 0000000..5dc3ce2 --- /dev/null +++ b/main.py @@ -0,0 +1,129 @@ +from fastapi import FastAPI, HTTPException +import uvicorn + +from database.database import Database + +app = FastAPI() +db = Database() + + +@app.get("/artnet/metadata/art") +async def art(id: int = None, hash: str = None, tag_id: int = None): + if id is None and hash is None and tag_id is None: + raise HTTPException(status_code=406, detail="requires id or hash param") + + result = None + if id is not None: + result = db.get_art_by_id(id) + elif hash is not None: + result = db.get_art_by_hash(hash) + elif tag_id is not None: + result = db.get_tag_art(tag_id) + + if result is not None: + return result + raise HTTPException(status_code=404, detail="Art was not found!") + + +@app.get("/artnet/metadata/presence") +async def presence(name: str = None, domain: str = None, artist_id: int = None, art_id: int = None, + art_md5: str = None): + if name is None and domain is None and artist_id is None and art_id is None and art_md5 is None: + raise HTTPException(status_code=406, detail="You must query with at least one parameter!") + + result = None + if artist_id is not None and name is None and domain is None: + result = db.get_artist_presences(artist_id) + + elif art_id is not None and name is None and domain is None: + result = db.get_art_presences_by_id(art_id) + + elif art_md5 is not None and name is None and domain is None: + result = db.get_art_presences_by_hash(art_md5) + + elif name is not None and domain is not None: + result = db.get_presence(name, domain) + + if result is not None: + return result + raise HTTPException(status_code=404, detail="Presence was not found!") + + +@app.get("/artnet/metadata/artist") +async def artist(id: int = None, topic_id: int = None): + if id is None and topic_id is None: + raise HTTPException(status_code=406) + + result = None + if id is not None: + result = db.get_artist(id) + elif topic_id is not None: + result = db.get_topic_artists(topic_id) + if result is not None: + return result + raise HTTPException(status_code=404, detail="Artist was not found!") + + +@app.get("/artnet/metadata/topic") +async def topic(name: str = None, id: int = None, artist_id: int = None): + if name is None and id is None and artist_id is None: + raise HTTPException(status_code=406) + + result = None + if name is not None or id is not None: + if id is not None: + result = db.get_topic_by_id(id) + elif name is not None: + result = db.get_topic_by_name(name) + + elif artist_id is not None: + result = db.get_artist_topics(artist_id) + + if result is not None: + return result + raise HTTPException(status_code=404, detail="Topic was not found!") + + +@app.get("/artnet/metadata/tag") +async def tag(id: int = None, name: str = None, fuzzy: bool = False, category: int = None): + if id is None and name is None and category is None: + raise HTTPException(status_code=406) + + result = None + if fuzzy: + if name is not None: + result = db.search_tag_by_name_fuzzy(name) + if result is not None: + return result + raise HTTPException(status_code=404, detail="No matching tag found!") + else: + if id is not None: + result = db.get_tag_by_id(id) + elif name is not None: + result = db.get_tag_by_name(name) + elif category is not None: + result = db.get_category_tags(category) + + if result is not None: + return result + raise HTTPException(status_code=404, detail="Tag was not found!") + + +@app.get("/artnet/metadata/category") +async def category(name: str = None, id: int = None): + if name is None and id is None: + raise HTTPException(status_code=406) + + result = None + if id is not None: + result = db.get_category_by_id(id) + elif name is not None: + result = db.get_category_by_name(name) + + if result is not None: + return result + raise HTTPException(status_code=404, detail="Category not found!") + + +if __name__ == "__main__": + uvicorn.run(app, host="127.0.0.1", port=8000)