SQLAlchemy kickstart tutorial

database.py

import datetime

from sqlalchemy.orm import declarative_base, sessionmaker, relationship, Session
from sqlalchemy import create_engine, Column, Integer, String, DateTime, ForeignKey

from app.exceptions import MarketException

engine = create_engine(
    "mysql+pymysql://root:rootroot@192.168.10.151:33061/Sample",
    isolation_level = 'REPEATABLE READ', echo = True
)

Base = declarative_base()


class Market(Base):
    __tablename__ = 'markets'

    id = Column(Integer, primary_key = True, autoincrement = "auto")
    market_name = Column(String(50), unique = True, nullable = False)
    created_at = Column(DateTime(), default = datetime.datetime.utcnow())

    # 역방향 relation
    stocks = relationship("Stock", back_populates = "market")

    def __repr__(self):
        return f"<Market(id={self.id} name={self.market_name})>"

    def __init__(self, name: str):
        self.market_name = self.clean_market_name(name.upper())

    def clean_market_name(self, name: str):
        if name not in ['KOSPI', 'KOSDAQ']:
            raise MarketException(f"Market({name}) is not allowed")

        return name


class Stock(Base):
    __tablename__ = 'stocks'

    id = Column(Integer, primary_key = True, autoincrement = "auto")

    # 정방향 relation
    market_id = Column(Integer, ForeignKey("markets.id"))
    market = relationship("Market", back_populates = "stocks")

    stock_code = Column(String(50), unique = True, nullable = False)
    stock_name = Column(String(50), unique = True, nullable = False, index = True)

    def __repr__(self):
        return f"<Stock(id={self.id} market={self.market_id} name={self.stock_name})>"

    def __init__(self, market_id: int, stock_code: str, stock_name: str):
        self.market_id = market_id
        self.stock_code = stock_code
        self.stock_name = stock_name


Base.metadata.create_all(engine)
Sess = sessionmaker(bind = engine)


def session_factory() -> Session:
    return Sess()

exceptions.py

class MarketException(Exception):
    pass

class MarketNotFound(MarketException):
    pass

class StockException(Exception):
    pass

class StockNotFound(StockException):
    pass

service.py

from sqlalchemy.exc import NoResultFound

from app.database import session_factory, Market, Stock
from app.exceptions import MarketNotFound


class MarketService:

    def create(self, market_name: str):
        with session_factory() as session:
            try:
                market = Market(market_name.upper())
                session.add(market)
                session.commit()
                raise Exception("123")

            except Exception:
                session.rollback()

    def get_or_create(self, market_name: str) -> Market:
        with session_factory() as session:
            try:
                market = session.query(Market).filter(Market.market_name == market_name.upper()).one()
                return market

            except NoResultFound:
                try:
                    market = Market(market_name)
                    session.add(market)
                    session.commit()
                    return market

                except Exception:
                    session.rollback()

    def delete_equal(self, market_name: str):
        with session_factory() as session:
            markets = session.query(Market).filter(Market.market_name == market_name.upper()).all()
            try:
                for market in markets:
                    session.delete(market)

            except Exception:
                session.rollback()
            session.commit()


class StockService:

    def create(self, market_name: str, stock_code: str, stock_name: str):
        with session_factory() as session:
            try:
                market = session.query(Market).filter(Market.market_name == market_name.upper()).one()

            except NoResultFound:
                try:
                    market = Market(market_name)
                    session.add(market)

                except Exception:
                    session.rollback()

            else:
                try:
                    stock = Stock(market.id, stock_code, stock_name)
                    session.add(stock)
                    session.commit()

                except Exception:
                    session.rollback()

    def delete_equal(self, stock_name: str):
        with session_factory() as session:
            stocks = session.query(Stock).filter(Stock.stock_name == stock_name).all()
            try:
                for stocks in stocks:
                    session.delete(stocks)

            except Exception:
                session.rollback()
            session.commit()