FrontPage/python/Flask

Sqlalchemy †

dbの定義

#! coding:utf-8
"""
使い方

	# Generate DB
	import db
	db.metadata.create_all()

"""

from sqlalchemy import create_engine, MetaData

# Generate Engine
engine = create_engine('sqlite:///db.sqlite3', echo=True)
metadata = MetaData() # まじない
metadata.bind = engine # まじない


## Define Table
from sqlalchemy import (Table, Column, Integer, String)
menus = Table(
	'menus', metadata,
	Column('id', Integer, primary_key=True),
	Column('name', String),
	Column('kcal', Integer)
	)

if __name__ == '__main__':
	# DB Create
	# python -c "import db;db.metadata.create_all()"
	metadata.create_all()

DBを使う(CRUB) †

#! coding:utf-8
"""
使い方

	
	# Tables
	menus = Table(
		'menus', metadata,
		Column('id', Integer, primary_key=True),
		Column('name', String),
		Column('kcal', Integer)
		)
	# CRUD(クラッド)
	Create, Read, Update, Delete
"""
import db

db.metadata.create_all()

# [C] Create
db.menus.insert().execute(name='チュッパチャプス', kcal=100)
db.menus.insert().execute(name='カレーメシ', kcal=500)
db.menus.insert().execute(name='ラーメン', kcal=700.1)

# [R] Read
data = db.menus.select().execute().fetchall()
for d in data:
	print(">> Select:{}".format(d))

# [R] Read
datas = db.menus.select().where(db.menus.c.kcal < 200).order_by('kcal').limit(5).execute().fetchall()
for d in datas:
	print(">> Select:{}".format(d))

# [R] Read
datas = db.menus.select().where(db.menus.c.kcal > 400).order_by('kcal').limit(5).execute().fetchall()
for d in datas:
	print(">> Select:{}".format(d))

# [U] Update
db.menus.update().where(db.menus.c.id == 1).execute(kcal=1000)

# [D] Delete
db.menus.delete().where(db.menus.c.kcal < 1000).execute()

SESSION †

models.py †

#! coding:utf-8
"""
"""

import unittest

from datetime import datetime
from sqlalchemy import create_engine
from sqlalchemy import (Column, Integer, String)
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    name = Column(String)
    fullname = Column(String)
    passward = Column(String)



if __name__ == '__main__':
    engine = create_engine('sqlite:///db.sqlite3', echo=True)
    Base.metadata.create_all(engine)
    
    unittest.main()

crud.py †

#! coding:utf-8
"""
"""

import unittest

from sqlalchemy import create_engine
from sqlalchemy.orm import sessionmaker
from model import User

# Generate Session
engine = create_engine('sqlite:///db.sqlite3', echo=True)
Session = sessionmaker(bind=engine)
session = Session()

# [D] DELETE ALL
session.query(User).delete()
session.commit()

# [C] Create
ed_user = User(name='ed', fullname='Ed Jones', passward='edspassword')
session.add(ed_user)

# [C] Create
session.add_all([
    User(name='wendy', fullname='Wendy Williams', passward="wpass"),
    User(name='mary', fullname='Mary Williams', passward="mmpass"),
    User(name='fred', fullname='Fred Williams', passward="fefepass"),
    ])

# Send commit - トランザクションをコミット
session.commit()

# [R] READ
for row in session.query(User).all():
    print(">>Read:",row.id, row.name, row.fullname, row.passward)

# [U] Update
row = session.query(User).filter_by(id=1).one()
row.name = "torina update"
session.add(row)
session.commit()

# [D] DELETE
row = session.query(User).filter_by(id=2).one()
session.delete(row)
session.commit()


# [R] READ
for row in session.query(User).all():
    print(">>Read:",row.id, row.name, row.fullname, row.passward)