python操作数据库 - SQLAlchemy

赤色烙印

　　ORM简介

　　ORM全称是Object>　　核心就是一个class对应一张表。
　　SQLAlchemy现在已经是python中最流行的ORM框架。
　　ORM优点：
　　1、隐藏数据库实现，让业务代码只访问对象而不是数据库表
　　2、良好的数据库操作接口，简单、学习成本低
　　3、动态数据表映射，表结构改变时，减少代码修改量
　　4、方便引入缓存功能
　　SQLAlchemy结构

　　SQLAlchemy ORM详解
yum install -y python-sqlalchemy # 或pip install sqlalchemy　　
yum install -y python-libs  # 安装sqlite3
　　
pip install sqlalchemy-utils # 安装sqlalchemy 一些工具类
　　sqlite数据库操作： http://blog.chinaunix.net/uid-26833883-id-3239313.html
　　创建数据库及表结构
　　[root@controller001 home]# cat sqlalchemy_tur.py
　　import sqlalchemy
　　from sqlalchemy import create_engine # operate database

　　from sqlalchemy.ext.declarative import declarative_base # create database table base>　　from sqlalchemy import Column, Integer, String # table structure type
　　from sqlalchemy.orm import sessionmaker # Use Database Transaction
　　# sqlite database is mostly used for android system.
　　# echo=True optios: print the real sql,mostly used in debug.
　　engine = create_engine('sqlite:///foo.db', echo=True)   # 创建数据库的引擎
　　# 如果是mysql engine，就要先创建database
　　[root@controller001 home]# cat sqlalchemy_tur_3.py
　　import sqlalchemy
　　from sqlalchemy import create_engine
　　from sqlalchemy_utils import database_exists, create_database # 需要安装sqlalchemy_utils
　　engine = create_engine('mysql://root:dbroot@localhost/sqlalchemy')
　　if not database_exists(engine.url):
　　create_database(engine.url)
　　print(database_exists(engine.url))
　　# 不同数据库的engine配置：http://sqlalchemy.readthedocs.org/en/rel_1_0/core/engines.html
　　Base = declarative_base()
　　class User(Base):
　　__tablename__ = 'users'
　　id = Column(Integer, primary_key=True)
　　name = Column(String)
　　fullname = Column(String)
　　password = Column(String)
　　def __repr__(self):
　　return "<User(name='%s', fullname='%s', password='%s')>" %(
　　self.name, self.fullname, self.password)
　　Base.metadata.create_all(engine)
　　执行后，会在本地生成一个foo.db sqlite数据库
　　插入数据
　　user_1 = User(name='user1', fullname='Yao', password='user1')
　　Session = sessionmaker(bind=engine)    # 使用数据库事务，保持数据原子性
　　session = Session()
　　session.add(user_1)
　　session.commit()
# 执行结果　　
[root@controller001 home]# python sqlalchemy_tur.py
　　
2015-09-13 20:59:16,178 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
　　
2015-09-13 20:59:16,179 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 20:59:16,179 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
　　
2015-09-13 20:59:16,179 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 20:59:16,181 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
　　
2015-09-13 20:59:16,182 INFO sqlalchemy.engine.base.Engine INSERT INTO users (name, fullname, password) VALUES (?, ?, ?)
　　
2015-09-13 20:59:16,182 INFO sqlalchemy.engine.base.Engine ('user1', 'Yao', 'user1')
　　
2015-09-13 20:59:16,184 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
　　
FROM users
　　
WHERE users.name = ?
　　
LIMIT ? OFFSET ?
　　
2015-09-13 20:59:16,185 INFO sqlalchemy.engine.base.Engine ('user1', 1, 0)
　　
2015-09-13 20:59:16,186 INFO sqlalchemy.engine.base.Engine COMMIT
　　
[root@controller001 home]# sqlite3 foo.db
　　
SQLite version 3.7.17 2013-05-20 00:56:22
　　
Enter ".help" for instructions
　　
Enter SQL statements terminated with a ";"
　　
sqlite> select * from users;
　　
1|user1|Yao|user1
　　插入多条数据
　　Session = sessionmaker(bind=engine)
　　session = Session()
　　session.add_all([
　　User(name='user1', fullname='user1_1', password='123'),
　　User(name='user2', fullname='user2_2', password='123'),
　　User(name='user3', fullname='user3_3', password='123')])
　　session.commit()
# 执行结果　　
root@controller001 home]# python sqlalchemy_tur.py
　　
2015-09-13 21:08:25,372 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
　　
2015-09-13 21:08:25,372 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 21:08:25,373 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
　　
2015-09-13 21:08:25,373 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 21:08:25,374 INFO sqlalchemy.engine.base.Engine BEGIN (implicit)
　　
2015-09-13 21:08:25,376 INFO sqlalchemy.engine.base.Engine INSERT INTO users (name, fullname, password) VALUES (?, ?, ?)
　　
2015-09-13 21:08:25,376 INFO sqlalchemy.engine.base.Engine ('user1', 'user1_1', '123')
　　
2015-09-13 21:08:25,377 INFO sqlalchemy.engine.base.Engine INSERT INTO users (name, fullname, password) VALUES (?, ?, ?)
　　
2015-09-13 21:08:25,377 INFO sqlalchemy.engine.base.Engine ('user2', 'user2_2', '123')
　　
2015-09-13 21:08:25,377 INFO sqlalchemy.engine.base.Engine INSERT INTO users (name, fullname, password) VALUES (?, ?, ?)
　　
2015-09-13 21:08:25,378 INFO sqlalchemy.engine.base.Engine ('user3', 'user3_3', '123')
　　
2015-09-13 21:08:25,380 INFO sqlalchemy.engine.base.Engine SELECT users.id AS users_id, users.name AS users_name, users.fullname AS users_fullname, users.password AS users_password
　　
FROM users
　　
WHERE users.name = ?
　　
LIMIT ? OFFSET ?
　　
2015-09-13 21:08:25,380 INFO sqlalchemy.engine.base.Engine ('user1', 1, 0)
　　
2015-09-13 21:08:25,381 INFO sqlalchemy.engine.base.Engine COMMIT
　　
[root@controller001 home]# sqlite3 foo.db
　　
SQLite version 3.7.17 2013-05-20 00:56:22
　　
Enter ".help" for instructions
　　
Enter SQL statements terminated with a ";"
　　
sqlite> select * from users;
　　
1|user1|Yao|user1
　　
2|user1|user1_1|123
　　
3|user2|user2_2|123
　　
4|user3|user3_3|123
　　查询
　　res = session.query(User).filter_by(name='user1').first()
　　res = session.query(User).all() # 全表扫描

　　for row in session.query(User).order_by(User.id): #>　　print row
　　for row in session.query(User).filter(User.name.in_(['yao','shen','tim'])):  # in操作
　　print row
　　for row in session.query(User).filter(~User.name.in_(['yao','shen','tim'])): # not in 操作
　　print row
　　session.query(User).filter(User.name == 'yao').count()  # 计数操作
　　from sqlalchemy import and_, or_
　　for row in session.query(User).filter(and_(User.name == 'yao', User.fullname=='yao')): # 与操作
　　print row
　　for row in session.query(User).filter(or_(User.name == 'yao', User.fullname=='yao')): # 或操作
　　print row
　　
　　创建一个带有外键的表
　　[root@controller001 home]# cat sqlalchemy_tur.py
　　import sqlalchemy
　　from sqlalchemy import create_engine # operate database

　　from sqlalchemy.ext.declarative import declarative_base # create database table base>　　from sqlalchemy import Column, Integer, String # table structure type
　　from sqlalchemy.orm import sessionmaker # Use Database Transaction
　　# sqlite database is mostly used for android system.
　　# echo=True optios: print the real sql,mostly used in debug.
　　engine = create_engine('sqlite:///foo.db', echo=True)
　　Base = declarative_base()
　　class User(Base):
　　__tablename__ = 'users'
　　id = Column(Integer, primary_key=True)
　　name = Column(String)
　　fullname = Column(String)
　　password = Column(String)
　　def __repr__(self):
　　return "<User(name='%s', fullname='%s', password='%s')>" %(
　　self.name, self.fullname, self.password)
　　from sqlalchemy import ForeignKey

　　from sqlalchemy.orm import>　　class Address(Base):
　　__tablename__ = 'address'
　　id = Column(Integer, primary_key=True)
　　email_address = Column(String, nullable=False)
　　user_id = Column(Integer, ForeignKey('users.id'))

　　user =>　　def __repr__(self):
　　return "<Address(email_address='%s')" % self.email_address
　　Base.metadata.create_all(engine)
# 执行结果　　
[root@controller001 home]# python sqlalchemy_tur.py
　　
2015-09-13 21:31:37,289 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
　　
2015-09-13 21:31:37,289 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 21:31:37,290 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
　　
2015-09-13 21:31:37,290 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 21:31:37,291 INFO sqlalchemy.engine.base.Engine PRAGMA table_info("users")
　　
2015-09-13 21:31:37,292 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 21:31:37,292 INFO sqlalchemy.engine.base.Engine PRAGMA table_info("address")
　　
2015-09-13 21:31:37,293 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 21:31:37,293 INFO sqlalchemy.engine.base.Engine
　　
CREATE TABLE address (
　　
id INTEGER NOT NULL,
　　
email_address VARCHAR NOT NULL,
　　
user_id INTEGER,
　　
PRIMARY KEY (id),
　　
FOREIGN KEY(user_id) REFERENCES users (id)
　　
)
　　
2015-09-13 21:31:37,293 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 21:31:37,343 INFO sqlalchemy.engine.base.Engine COMMIT
　　插入数据
　　ice = User(name='ice', fullname='ice yao', password='ice')
　　ice.addresses = [
　　Address(email_address='ice@gg.com'),
　　Address(email_address='yao@gg.com')
　　]
　　session.add(ice)
　　session.commit()
　　for u, a in session.query(User, Address).\
　　filter(User.id==Address.user_id).\
　　filter(Address.email_address=='ice@gg.com').\
　　all():
　　print u, a
　　SQLAlchemy Core详解
　　
　　创建数据库及表
　　[root@controller001 home]# cat sqlalchemy_tur_2.py
　　import sqlalchemy
　　from sqlalchemy import create_engine
　　from sqlalchemy import Table, Column, Integer, String, MetaData, ForeignKey
　　engine = create_engine('sqlite:///foo.db', echo=True)
　　metadata = MetaData()
　　users = Table('users', metadata,         # 这里使用Table来创建表
　　Column('id', Integer, primary_key=True),
　　Column('name', String),
　　Column('fullname', String)
　　)
　　address = Table('address', metadata,
　　Column('id', Integer, primary_key=True),
　　Column('user_id', None, ForeignKey('users.id')),
　　Column('email_address', String, nullable=False)
　　)
　　metadata.create_all(engine)
　　
　　查看sqlite数据库
　　
　　sqlite> .schema
　　CREATE TABLE users (
　　id INTEGER NOT NULL,
　　name VARCHAR,
　　fullname VARCHAR,
　　PRIMARY KEY (id)
　　);
　　CREATE TABLE address (
　　id INTEGER NOT NULL,
　　user_id INTEGER,
　　email_address VARCHAR NOT NULL,
　　PRIMARY KEY (id),
　　FOREIGN KEY(user_id) REFERENCES users (id)
　　);
　　插入数据
conn = engine.connect()   # 初始化一个连接　　
conn.execute(users.insert(), [dict(name='jack', fullname='jack jons'),
　　
dict(name='wendy', fullname='wendy ice')])
　　
　　查询
　　from sqlalchemy.sql import select
　　s = select([users])    # 语法很像sql
　　res = conn.execute(s)
　　for row in res:
　　print row
# 执行结果　　
[root@controller001 home]# python sqlalchemy_tur_2.py
　　
2015-09-13 22:24:14,459 INFO sqlalchemy.engine.base.Engine SELECT CAST('test plain returns' AS VARCHAR(60)) AS anon_1
　　
2015-09-13 22:24:14,460 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 22:24:14,460 INFO sqlalchemy.engine.base.Engine SELECT CAST('test unicode returns' AS VARCHAR(60)) AS anon_1
　　
2015-09-13 22:24:14,460 INFO sqlalchemy.engine.base.Engine ()
　　
2015-09-13 22:24:14,462 INFO sqlalchemy.engine.base.Engine SELECT users.id, users.name, users.fullname
　　
FROM users
　　
2015-09-13 22:24:14,462 INFO sqlalchemy.engine.base.Engine ()
　　
(1, u'jack', u'jack jons')
　　
(2, u'wendy', u'wendy ice')
　　# 带where的条件查询
　　from sqlalchemy.sql import select
　　s = select([users, address]).where(users.c.id == address.c.user_id)
　　res = conn.execute(s)
　　for row in res:
　　print row
　　# 使用text,更像sql了
　　from sqlalchemy.sql import select
　　s = text(

　　"SELECT users.fullname || ',' || address.email_address AS>　　"FROM users, address "
　　"WHERE users.id = address.user_id "
　　"AND users.name BETWEEN :x AND :y "
　　"AND (address.email_address LIKE :t1 "
　　"OR address.email_address LIKE :t2)")
　　res = conn.execute(s, x='m',y='z',t1='@yao.com',t2='@ice.com').fetchall()
　　for row in res:
　　print row