recipe-graph/src/db.py

65 lines
2.0 KiB
Python

from sqlalchemy import create_engine, Column, Integer, String, \
ForeignKey, UniqueConstraint
from sqlalchemy.engine import URL
from sqlalchemy.ext.declarative import declarative_base
import os
from dotenv import load_dotenv
import logging
Base = declarative_base()
class Ingredient(Base):
__tablename__ = 'Ingredient'
id = Column(Integer, primary_key = True)
name = Column(String, nullable = False)
class RecipeSite(Base):
__tablename__ = 'RecipeSite'
id = Column(Integer, primary_key = True)
name = Column(String, nullable = False, unique = True)
ingredient_class = Column(String, nullable = False)
name_class = Column(String, nullable = False)
base_url = Column(String, nullable = False, unique = True)
class Recipe(Base):
__tablename__ = 'Recipe'
id = Column(Integer, primary_key = True)
name = Column(String)
identifier = Column(String, nullable = False)
recipe_site_id = Column(Integer, ForeignKey('RecipeSite.id'))
UniqueConstraint(identifier, recipe_site_id)
class RecipeIngredient(Base):
__tablename__ = 'RecipeIngredient'
id = Column(Integer, primary_key = True)
text = Column(String, nullable = False)
recipe_id = Column(Integer, ForeignKey('Recipe.id'))
ingredient_id = Column(Integer, ForeignKey("Ingredient.id"))
def get_engine(use_dotenv = True, **kargs):
if use_dotenv:
load_dotenv()
DB_URL = os.getenv("POSTGRES_URL")
DB_USER = os.getenv("POSTGRES_USER")
DB_PASSWORD = os.getenv("POSTGRES_PASSWORD")
DB_NAME = os.getenv("POSTGRES_DB")
eng_url = URL.create('postgresql',
username=DB_USER,
password=DB_PASSWORD,
host=DB_URL,
database=DB_NAME)
return create_engine(eng_url)
if __name__ == "__main__":
eng = get_engine()
logging.info(f"Createing DB Tables: {eng.url}")
Base.metadata.create_all(eng, checkfirst=True)