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
创建数据库及表结构
# 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
# 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()
# 执行结果
# 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
# 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
# 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
创建一个带有外键的表
# 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)
# 执行结果
# 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详解
创建数据库及表
# 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() # 语法很像sql
res = conn.execute(s)
for row in res:
print row
# 执行结果
# 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().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
页:
[1]