ORM (SQL Alchemy)

 

ORM (SQL Alchemy)

Review

 

 

---

 

1. Create Tables

realationship : class 간 연결

• backref : 상호간 연결
  Indicates the string name of a property to be placed on the related mapper’s class that will handle this relationship in the other direction. The other property will be created automatically when the mappers are configured. Can also be passed as a backref() object to control the configuration of the new relationship.
• back_populates : 일방향 연결
  Takes a string name and has the same meaning as relationship.backref, except the complementing property is not created automatically, and instead must be configured explicitly on the other mapper. The complementing property should also indicate relationship.back_populates to this relationship to ensure proper functioning.
• cascade : 연속 기능
  A comma-separated list of cascade rules which determines how Session operations should be “cascaded” from parent to child. This defaults to False, which means the default cascade should be used - this default cascade is "save-update, merge".
  The available cascades are save-update, merge, expunge, delete, delete-orphan, and refresh-expire. An additional option, all indicates shorthand for "save-update, merge, refresh-expire, expunge, delete", and is often used as in "all, delete-orphan" to indicate that related objects should follow along with the parent object in all cases, and be deleted when de-associated.

from sqlalchemy import create_engine, Column, Integer, String, ForeignKey,\
    DateTime, VARCHAR
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import relationship, backref

## 기본설정
DATABASE_URI = 'sqlite:///orm.db'
engine = create_engine(DATABASE_URI)
Base = declarative_base() # 매핑을 위한 클래스, 아래에서 상속을 통해 매핑

## 초기화
def reset_db(engine=engine):
    Base.metadata.drop_all(bind=engine)
    Base.metadata.create_all(bind=engine)

## 클래스 생성 (테이블)
class Address(Base):
    __tablename__ = "Address"
    address_id = Column(Integer,nullable=False, primary_key=True)
    city = Column(VARCHAR(48))
    street = Column(VARCHAR(128))
    customer_id = Column(Integer,ForeignKey('Customer.customer_id'))
    
    cusotmers = relationship("Customer", backref='address', cascade="all, delete")

class Item(Base):
    __tablename__ = "Item"
    item_id = Column(Integer,nullable=False, primary_key=True)
    item_name = Column(VARCHAR(32),nullable=False)
    item_price = Column(Integer)


class Invoice_Item(Base):
    __tablename__ = "Invoice_Item"
    ii_id = Column(Integer,nullable=False, primary_key=True)
    invoice_date = Column(DateTime,nullable=False)
    address_id = Column(Integer,ForeignKey('Address.address_id'))
    item_id = Column(Integer,ForeignKey('Item.item_id'))
 
    addresses = relationship("Address", backref='invoice_item', cascade="all, delete")
    items = relationship("Item", backref='invoice_item', cascade="all, delete")


class Customer(Base):
    __tablename__ = "Customer"
    customer_id = Column(Integer,nullable=False, primary_key=True)
    customer_name = Column(VARCHAR(32),nullable=False)
    customer_age = Column(Integer)

 

 

2. How to Use

## Insert
def insert_address(customer_id, address_id=None, city='', street='',
                   session=session):
    inserting = Address(customer_id=customer_id, address_id=address_id, city=city, street=street)
    session.add(inserting)
    
    
## Return
def read_item_invoices_date(item_id, session=session):
    result = session.query(Invoice_Item.invoice_date)\
                    .filter(Invoice_Item.ii_id == item_id).all()
    return result

## JOIN
def read_customer_item_names(customer_id, session=session):
    result = session.query(Item.item_name)\
                    .join(Invoice_Item).join(Address)\ # Item과 연결된 Invoce_Item먼저 Join /그 뒤에 Invoice_Item과 연결된 Address JOIN
                    .filter(Address.customer_id == customer_id).all()
    return result


## Delete
def delete_customer(customer_id, session=session):
    to_delete = session.query(Customer)\
                .filter(Customer.customer_id == customer_id).one()
    session.delete(to_delete)
    
## Update
def update_item_price(item_id, new_price, session=session):
    to_update = session.query(Item)\
                .filter(Item.item_id == item_id).first()
    to_update.item_price = new_price
    session.commit()

 

 

3. Use

## 패키지
from csv import DictReader
from sqlalchemy import create_engine, Column, Integer, String, Float
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import Session

## DB및 파일 지정
DATABASE_URI = "sqlite:///titanic_orm.db"
CSV_FILEPATH = 'titanic.txt'

## session 만들기
engine = create_engine(DATABASE_URI)
Base = declarative_base(bind=engine)
session = Session(bind=engine)

## 객체 생성
class Passenger(Base):
    __tablename__ = "Passenger"
    ID = Column(Integer,nullable=False, primary_key=True)
    Survived = Column(Integer,nullable=False)
    Pclass = Column(Integer,nullable=False)
    Name = Column(String,nullable=False)
    Sex = Column(String,nullable=False)
    Age = Column(Float,nullable=False)
    Siblings_Spouses_Aboard = Column(Integer,nullable=False)
    Parents_Children_Aboard = Column(Integer,nullable=False)
    Fare = Column(Float,nullable=False)

## csv 함수 생성
def read_csv(csvfile,header,assign_header) : 
    reader = DictReader(csvfile)
    rows = []
    for each in reader:
        row={}
        for field,field_toassign in zip(header,assign_header):
            row[field_toassign]=each[field]
        rows.append(row)
    return rows
    
    
## insert 함수 생성 (for dict)
def insert_data(rows,session=session):

    for row in rows :
        Survived = row['Survived']
        Pclass = row['Pclass']
        Name = row['Name']
        Sex = row['Sex']
        Age = row['Age']
        Siblings_Spouses_Aboard = row['Siblings_Spouses_Aboard']
        Parents_Children_Aboard = row['Parents_Children_Aboard']
        Fare = row['Fare']

        inserting = Passenger(
            Survived = Survived,
            Pclass = Pclass,
            Name = Name,
            Sex = Sex,
            Age = Age,
            Siblings_Spouses_Aboard = Siblings_Spouses_Aboard,
            Parents_Children_Aboard = Parents_Children_Aboard,
            Fare = Fare
        )
        session.add(inserting)

## Add
csvfile = open(r"C:\Users\aodl3\Desktop\n324\ds-sa-orm\titanic.txt")
reader = DictReader(csvfile)
header = ["Survived","Pclass","Name","Sex","Age","Siblings/Spouses Aboard","Parents/Children Aboard","Fare"]
assign_header = ["Survived","Pclass","Name","Sex","Age","Siblings_Spouses_Aboard","Parents_Children_Aboard","Fare"]

rows = read_csv(csvfile,header,assign_header)
insert_data(rows)

 

 

 

www.sqlalchemy.org/

SQLAlchemy - The Database Toolkit for Python