テーブル作成とデータの挿入

SQLAlchemy では、テーブルを定義する様々な方法がありますが、VIEW の作成も統一して記述したいので、SQL文をそのまま記述する方法を使うことにします。

db_construct.py:

#!/usr/bin/env python3

from sqlalchemy import *

def execute(engine, statements):
    c = engine.connect()
    try:
        tr = c.begin()
        try:
            for s in statements:
                c.execute(s)
            tr.commit()
        except:
            tr.rollback()
            raise
    finally:
        c.close()

def execute_list(engine, statements_list):
    for l in statements_list:
        execute(engine, l)

engine = create_engine("sqlite:///crm.db", echo=True)
engine.execute("PRAGMA foreign_keys = true")

statements = [
    ("CREATE TABLE person ("
         "id INTEGER PRIMARY KEY AUTOINCREMENT UNIQUE NOT NULL, "
         "name TEXT NOT NULL, "
         "customer_id INTEGER, "
         "FOREIGN KEY(customer_id) REFERENCES customer(id))"),
    ("CREATE TABLE company ("
         "id INTEGER PRIMARY KEY AUTOINCREMENT UNIQUE NOT NULL, "
         "name TEXT NOT NULL, "
         "customer_id INTEGER, "
         "FOREIGN KEY(customer_id) REFERENCES customer(id))"),
    ("CREATE TABLE customer ("
         "id INTEGER PRIMARY KEY AUTOINCREMENT UNIQUE NOT NULL)")
    ]

def execute_all():
    execute_list(engine, [
        statements
        ])

実験を兼ねてるので変った実装になっているのではないかと。

続いて、実験用のメインルーチン。

db_test.py:

#!/usr/bin/env python3

from sqlalchemy import *
from sqlalchemy.orm import sessionmaker
from sqlalchemy.orm import mapper
from sqlalchemy.orm import relation
from db_construct import *

execute_all()

metadata = MetaData()
metadata.reflect(bind=engine)

class Customer:
    pass

class Person:
    def __init__(self, name):
        self.name = name

class Company:
    def __init__(self, name):
        self.name = name

mapper(Person, metadata.tables['person'], properties={
    'customer': relation(Customer, uselist=False, backref='person')
})
mapper(Company, metadata.tables['company'], properties={
    'customer': relation(Customer, uselist=False, backref='company')
})
mapper(Customer, metadata.tables['customer'])

def add_person(session, name):
    customer = Customer()
    person = Person(name)
    person.customer = customer
    session.add(person)
    
def add_company(session, name):
    customer = Customer()
    company = Company(name)
    company.customer = customer
    session.add(company)

Session = sessionmaker(bind=engine)
session = Session()
add_person(session, "鈴木一郎")
add_person(session, "鈴木次郎")
add_company(session, "株式会社鈴木")
session.commit()

for i in session.query(Person).order_by(Person.id):
    print(i.id, i.name, i.customer_id)

for i in session.query(Company).order_by(Company.id):
    print(i.id, i.name, i.customer_id)

for i in session.query(Customer).order_by(Customer.id):
    print(i.id)

session.close()

Python的に書き方がよろしくなかったらご指摘くださいませ。