74 lines
No EOL
1.9 KiB
Python
74 lines
No EOL
1.9 KiB
Python
import os
|
|
import logging
|
|
from typing import Tuple, Dict
|
|
|
|
from sqlalchemy.orm import sessionmaker
|
|
from sqlalchemy import Engine, create_engine
|
|
|
|
from backend.db.pgclass import Base, articles
|
|
|
|
dblogger = logging.getLogger("backend.db")
|
|
dblogger.setLevel(level=logging.DEBUG)
|
|
|
|
db:Engine = None
|
|
|
|
def init():
|
|
global db
|
|
try:
|
|
dbpath = os.path.abspath("./backend/db/id2igid.db")
|
|
db = create_engine(f"sqlite:///{dbpath}")
|
|
Base.metadata.create_all(db)
|
|
except:
|
|
dblogger.critical("Cannot connect to database")
|
|
raise Exception("Cannot connect to database id2igid.db")
|
|
|
|
|
|
def get_session():
|
|
Session = sessionmaker(bind=db)
|
|
return Session()
|
|
|
|
|
|
def solo_article_fetcher(aid:int=None, igid:str=None) -> Dict | None:
|
|
with get_session() as session:
|
|
# query
|
|
if aid is not None: # has aid
|
|
res = session.query(articles).filter(articles.id == aid).first()
|
|
elif igid is not None: # no aid , has igid
|
|
res = session.query(articles).filter(articles.igid == igid).first()
|
|
else:
|
|
return None
|
|
|
|
# process result
|
|
if res is None:
|
|
return None
|
|
else:
|
|
return {"id": res.id, "igid": res.igid}
|
|
|
|
|
|
def solo_article_inserter(aid:int, igid:str) -> int: # TODO
|
|
with get_session() as session:
|
|
# check if exists
|
|
res = session.query(articles).filter(articles.id == aid).first()
|
|
if res is not None:
|
|
return 1
|
|
|
|
# insert
|
|
new_article = articles(id=aid, igid=igid)
|
|
session.add(new_article)
|
|
session.commit()
|
|
|
|
return 0
|
|
|
|
|
|
def solo_article_remover(aid:int) -> int:
|
|
with get_session() as session:
|
|
# check if exists
|
|
res = session.query(articles).filter(articles.id == aid).first()
|
|
if res is None:
|
|
return 1
|
|
|
|
# delete
|
|
session.delete(res)
|
|
session.commit()
|
|
|
|
return 0 |