copywriting_production/src/manager/core/db.py

from datetime import datetime
from typing import Optional
from sqlmodel import SQLModel, create_engine, Session, select, Field
from config.settings import DB_URL
from utils.sql_engine import create_db_and_tables,drop_table,engine
from src.models.asin_model import AsinSeed

class DbManager:
    def __init__(self, engine: str=None):
        self.engine = engine or create_engine(DB_URL)
        create_db_and_tables()

    def save_asin_seed(self, asin_model: AsinSeed):
        with Session(self.engine) as session:
            session.add(asin_model)
            session.commit()
            session.refresh(asin_model)
            return asin_model

    def get_asin_seed(self, asin: str)->AsinSeed:
        with Session(self.engine) as session:
            statement = select(AsinSeed).where(AsinSeed.asin == asin)
            results = session.exec(statement)
            return results.first()
    
    def add_or_ignore_asin_seed(self, asin_model: AsinSeed):
        exist = self.get_asin_seed(asin_model.asin)
        if exist:
            return exist
        else:
            return self.save_asin_seed(asin_model)
               
def main():
    asinseed_list = ['B0CQ1SHD8V', 'B0B658JC22', 'B0DQ84H883', 'B0D44RT8R8']
    db_manager = DbManager()
    for asin in asinseed_list:
        db_manager.add_or_ignore_asin_seed(AsinSeed(asin=asin, asin_area='JP'))


if __name__ == "__main__":
    main()