SQLAlchemy

from typing import List
from typing import Optional
from sqlalchemy import ForeignKey
from sqlalchemy import String
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship

class Base(DeclarativeBase):
    pass

class User(Base):
    __tablename__ = "user_account"
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]]
    addresses: Mapped[List["Address"]] = relationship(
        back_populates="user", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str]
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")
    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"

from sqlalchemy import create_engine

engine = create_engine("sqlite://", echo=True)

Base.metadata.create_all(engine)

from sqlalchemy.orm import Session

with Session(engine) as session:
    spongebob = User(
        name="spongebob",
        fullname="Spongebob Squarepants",
        addresses=[Address(email_address="[email protected]")],
    )
    sandy = User(
        name="sandy",
        fullname="Sandy Cheeks",
        addresses=[
            Address(email_address="[email protected]"),
            Address(email_address="[email protected]"),
        ],
    )
    patrick = User(name="patrick", fullname="Patrick Star")
    session.add_all([spongebob, sandy, patrick])
    session.commit()

from sqlalchemy import select

session = Session(engine)

stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))

for user in session.scalars(stmt):
    print(user)

stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "sandy")
    .where(Address.email_address == "[email protected]")
)
sandy_address = session.scalars(stmt).one()
print(sandy_address)

from sqlalchemy.ext.declarative import declarative_base

# 기본 클래스를 생성
Base = declarative_base()

# 사용자 테이블을 나타내는 클래스 정의
class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    email = Column(String)

from sqlalchemy import create_engine
engine = create_engine('mssql+pymssql://username:passwd@host/database', echo=True)

from sqlalchemy.ext.declarative import declarative_base

# 기본 클래스를 생성
Base = declarative_base()

# 사용자 테이블을 나타내는 클래스 정의
class TbEpAiscrapchargeIfTable(Base):
    __tablename__ = "TB_EP_AISCRAPCHARGE_IF"
    __table_args__ = {"extend_existing": True}

    idx = Column(Integer, primary_key=True)
    # ...

from sqlalchemy import Column, DateTime, func

class TbEpAiscrapchargeIfTable(Base, BaseTable):
    __tablename__ = "TB_EP_AISCRAPCHARGE_IF"
    __table_args__ = {"extend_existing": True}

    insert_date = Column("INSERT_DATE", DateTime, default=func.now())

from datetime import datetime
from sqlalchemy.orm import Session
from typing import List

def get_rows_after_date(session: Session, target_date: datetime) -> List[TbEpAiscrapchargeIfTable]:
    """
    특정 날짜 이후 insert_date 컬럼에 저장된 행을 가져오고, 날짜 기준으로 오름차순 정렬하는 함수.

    Parameters:
    - session: SQLAlchemy 세션 객체
    - target_date: datetime 객체로 특정 날짜

    Returns:
    - List[TbEpAiscrapchargeIfTable]: 해당 날짜 이후의 모든 행 목록 (오름차순 정렬)
    """
    return (session.query(TbEpAiscrapchargeIfTable)
            .filter(TbEpAiscrapchargeIfTable.insert_date > target_date)
            .order_by(TbEpAiscrapchargeIfTable.insert_date.asc())
            .all())

def all_heat_infos():
    with DBManager.instance().query(TbEpAiscrapchargeIfTable) as q:
        return q.all()

all_heat_infos().idx  # Error!!

def all_heat_infos():
    with DBManager.instance().query(TbEpAiscrapchargeIfTable) as q:
        rows = q.all()
        q.session.expunge_all()
        return rows

all_heat_infos().idx  # OK

def get_heat_infos_after_seq(if_seq: int):
    with DBManager.instance().query(TbEpAiscrapchargeIfTable) as q:
        rows = (
            q.filter(TbEpAiscrapchargeIfTable.if_seq >= if_seq)
            .order_by(asc(TbEpAiscrapchargeIfTable.if_seq))
            .all()
        )
        q.session.expunge_all()
        return rows

from datetime import datetime
from sqlalchemy.orm import Session
from sqlalchemy import and_

def delete_rows_before_date(cutoff_date: datetime, session: Session):
    """
    주어진 기준일 이전의 모든 행을 삭제합니다.

    Parameters:
        cutoff_date (datetime): 기준일
        session (Session): SQLAlchemy 세션 객체
    """
    # 기준일 이전의 모든 행을 조회하여 삭제
    session.query(HeatInfoCandidateTable) \
           .filter(HeatInfoCandidateTable.reg_ddtt < cutoff_date) \
           .delete(synchronize_session=False)
    session.commit()

# 사용 예시
with DBManager.instance().Session.begin() as session:
    cutoff_date = datetime(2023, 1, 1)  # 예: 2023년 1월 1일 이전의 데이터 삭제
    delete_rows_before_date(cutoff_date, session)

@staticmethod
def select_charging_info_with_heat_text(heat_no: int):
    with DBManager.instance().query(ChargingInfoTable) as q:
        rows = q.filter(ChargingInfoTable.heat_text.like(f"D_{heat_no}")).all()
        q.session.expunge_all()
        return rows

from urllib.parse import quote_plus

password = quote_plus("PassW@rd")

class TbEpAiscrapchargeIfTable(Base, BaseTable):
    __tablename__ = "TB_EP_AISCRAPCHARGE_IF"
    __table_args__ = {"extend_existing": True}

    if_seq = Column("IF_SEQ", Integer, nullable=True),  # 여기서 마지막 쉼표 빼자.