python sqlalchemy 多线程怎么写

python sqlalchemy 多线程怎么写,第1张

首先说下,由于最新的 0.8 版还是开发版本,因此我使用的是 0.79 版,API 也许会有些不同。

因为我是搭配 MySQL InnoDB 使用,所以使用其他数据库的也不能完全照搬本文。

接着就从安装开始介绍吧,以 Debian/Ubuntu 为例(请确保有管理员权限):

1.MySQL

复制代码代码如下:

apt-get install mysql-server

apt-get install mysql-client

apt-get install libmysqlclient15-dev

2.python-mysqldb

复制代码代码如下:

apt-get install python-mysqldb

3.easy_install

python ez_setup.py

4.MySQL-Python

复制代码代码如下:

easy_install MySQL-Python

5.SQLAlchemy

复制代码代码如下:

easy_install SQLAlchemy

如果是用其他 *** 作系统,遇到问题就 Google 一下吧。我是在 Mac OS X 上开发的,途中也遇到些问题,不过当时没记下来……

值得一提的是我用了 MySQL-Python 来连 MySQL,因为不支持异步调用,所以和 Tornado 不是很搭。不过性能其实很好,因此以后再去研究下其他方案吧……

装好后就可以开始使用了:

复制代码代码如下:

from sqlalchemy import create_engine

from sqlalchemy.orm import sessionmaker

DB_CONNECT_STRING = 'mysql+mysqldb://root:123@localhost/ooxx?charset=utf8'

engine = create_engine(DB_CONNECT_STRING, echo=True)

DB_Session = sessionmaker(bind=engine)

session = DB_Session()

这里的 DB_CONNECT_STRING 就是连接数据库的路径。“mysql+mysqldb”指定了使用 MySQL-Python 来连接,“root”和“123”分别是用户名和密码,“localhost”是数据库的域名,“ooxx”是使用的数据库名(可省略),“charset”指定了连接时使用的字符集(可省略)。

create_engine() 会返回一个数据库引擎,echo 参数为 True 时,会显示每条执行的 SQL 语句,生产环境下可关闭。

sessionmaker() 会生成一个数据库会话类。这个类的实例可以当成一个数据库连接,它同时还记录了一些查询的数据,并决定什么时候执行 SQL 语句。由于 SQLAlchemy 自己维护了一个数据库连接池(默认 5 个连接),因此初始化一个会话的开销并不大。对 Tornado 而言,可以在 BaseHandler 的 initialize() 里初始化:

复制代码代码如下:

class BaseHandler(tornado.web.RequestHandler):

def initialize(self):

self.session = models.DB_Session()

def on_finish(self):

self.session.close()

对其他 Web 服务器来说,可以使用 sqlalchemy.orm.scoped_session,它能保证每个线程获得的 session 对象都是唯一的。不过 Tornado 本身就是单线程的,如果使用了异步方式,就可能会出现问题,因此我并没使用它。

拿到 session 后,就可以执行 SQL 了:

复制代码代码如下:

session.execute('create database abc')

print session.execute('show databases').fetchall()

session.execute('use abc')

# 建 user 表的过程略

print session.execute('select * from user where id = 1').first()

print session.execute('select * from user where id = :id', {'id': 1}).first()

不过这和直接使用 MySQL-Python 没啥区别,所以就不介绍了;我还是喜欢 ORM 的方式,这也是我采用 SQLAlchemy 的唯一原因。

于是来定义一个表:

复制代码代码如下:

from sqlalchemy import Column

from sqlalchemy.types import CHAR, Integer, String

from sqlalchemy.ext.declarative import declarative_base

BaseModel = declarative_base()

def init_db():

BaseModel.metadata.create_all(engine)

def drop_db():

BaseModel.metadata.drop_all(engine)

class User(BaseModel):

__tablename__ = 'user'

id = Column(Integer, primary_key=True)

name = Column(CHAR(30)) # or Column(String(30))

init_db()

declarative_base() 创建了一个 BaseModel 类,这个类的子类可以自动与一个表关联。

以 User 类为例,它的 __tablename__ 属性就是数据库中该表的名称,它有 id 和 name 这两个字段,分别为整型和 30 个定长字符。Column 还有一些其他的参数,我就不解释了。

最后,BaseModel.metadata.create_all(engine) 会找到 BaseModel 的所有子类,并在数据库中建立这些表;drop_all() 则是删除这些表。

接着就开始使用这个表吧:

复制代码代码如下:

from sqlalchemy import func, or_, not_

user = User(name='a')

session.add(user)

user = User(name='b')

session.add(user)

user = User(name='a')

session.add(user)

user = User()

session.add(user)

session.commit()

query = session.query(User)

print query # 显示SQL 语句

print query.statement # 同上

for user in query: # 遍历时查询

print user.name

print query.all() # 返回的是一个类似列表的对象

print query.first().name # 记录不存在时,first() 会返回 None

# print query.one().name # 不存在,或有多行记录时会抛出异常

print query.filter(User.id == 2).first().name

print query.get(2).name # 以主键获取,等效于上句

print query.filter('id = 2').first().name # 支持字符串

query2 = session.query(User.name)

print query2.all() # 每行是个元组

print query2.limit(1).all() # 最多返回 1 条记录

print query2.offset(1).all() # 从第 2 条记录开始返回

print query2.order_by(User.name).all()

print query2.order_by('name').all()

print query2.order_by(User.name.desc()).all()

print query2.order_by('name desc').all()

print session.query(User.id).order_by(User.name.desc(), User.id).all()

print query2.filter(User.id == 1).scalar() # 如果有记录,返回第一条记录的第一个元素

print session.query('id').select_from(User).filter('id = 1').scalar()

print query2.filter(User.id >1, User.name != 'a').scalar() # and

query3 = query2.filter(User.id >1) # 多次拼接的 filter 也是 and

query3 = query3.filter(User.name != 'a')

print query3.scalar()

print query2.filter(or_(User.id == 1, User.id == 2)).all() # or

print query2.filter(User.id.in_((1, 2))).all() # in

query4 = session.query(User.id)

print query4.filter(User.name == None).scalar()

print query4.filter('name is null').scalar()

print query4.filter(not_(User.name == None)).all() # not

print query4.filter(User.name != None).all()

print query4.count()

print session.query(func.count('*')).select_from(User).scalar()

print session.query(func.count('1')).select_from(User).scalar()

print session.query(func.count(User.id)).scalar()

print session.query(func.count('*')).filter(User.id >0).scalar() # filter() 中包含 User,因此不需要指定表

print session.query(func.count('*')).filter(User.name == 'a').limit(1).scalar() == 1 # 可以用 limit() 限制 count() 的返回数

print session.query(func.sum(User.id)).scalar()

print session.query(func.now()).scalar() # func 后可以跟任意函数名,只要该数据库支持

print session.query(func.current_timestamp()).scalar()

print session.query(func.md5(User.name)).filter(User.id == 1).scalar()

query.filter(User.id == 1).update({User.name: 'c'})

user = query.get(1)

print user.name

user.name = 'd'

session.flush() # 写数据库,但并不提交

print query.get(1).name

session.delete(user)

session.flush()

print query.get(1)

session.rollback()

print query.get(1).name

query.filter(User.id == 1).delete()

session.commit()

print query.get(1)

增删改查都涉及到了,自己看看输出的 SQL 语句就知道了,于是基础知识就介绍到此了。

下面开始介绍一些进阶的知识。

如何批量插入大批数据?

可以使用非 ORM 的方式:

复制代码代码如下:

session.execute(

User.__table__.insert(),

[{'name': `randint(1, 100)`,'age': randint(1, 100)} for i in xrange(10000)]

)

session.commit()

上面我批量插入了 10000 条记录,半秒内就执行完了;而 ORM 方式会花掉很长时间。

如何让执行的 SQL 语句增加前缀?

使用 query 对象的 prefix_with() 方法:

复制代码代码如下:

session.query(User.name).prefix_with('HIGH_PRIORITY').all()

session.execute(User.__table__.insert().prefix_with('IGNORE'), {'id': 1, 'name': '1'})

如何替换一个已有主键的记录?

使用 session.merge() 方法替代 session.add(),其实就是 SELECT + UPDATE:

复制代码代码如下:

user = User(id=1, name='ooxx')

session.merge(user)

session.commit()

或者使用 MySQL 的 INSERT … ON DUPLICATE KEY UPDATE,需要用到 @compiles 装饰器,有点难懂,自己搜索看吧:《SQLAlchemy ON DUPLICATE KEY UPDATE》 和 sqlalchemy_mysql_ext。

如何使用无符号整数?

可以使用 MySQL 的方言:

复制代码代码如下:

from sqlalchemy.dialects.mysql import INTEGER

id = Column(INTEGER(unsigned=True), primary_key=True)

模型的属性名需要和表的字段名不一样怎么办?

开发时遇到过一个奇怪的需求,有个其他系统的表里包含了一个“from”字段,这在 Python 里是关键字,于是只能这样处理了:

复制代码代码如下:

from_ = Column('from', CHAR(10))

如何获取字段的长度?

Column 会生成一个很复杂的对象,想获取长度比较麻烦,这里以 User.name 为例:

复制代码代码如下:

User.name.property.columns[0].type.length

如何指定使用 InnoDB,以及使用 UTF-8 编码?

最简单的方式就是修改数据库的默认配置。如果非要在代码里指定的话,可以这样:

复制代码代码如下:

class User(BaseModel):

__table_args__ = {

'mysql_engine': 'InnoDB',

'mysql_charset': 'utf8'

}

MySQL 5.5 开始支持存储 4 字节的 UTF-8 编码的字符了,iOS 里自带的 emoji(如 �� 字符)就属于这种。

如果是对表来设置的话,可以把上面代码中的 utf8 改成 utf8mb4,DB_CONNECT_STRING 里的 charset 也这样更改。

如果对库或字段来设置,则还是自己写 SQL 语句比较方便,具体细节可参考《How to support full Unicode in MySQL databases》。

不建议全用 utf8mb4 代替 utf8,因为前者更慢,索引会占用更多空间。

如何设置外键约束?

复制代码代码如下:

from random import randint

from sqlalchemy import ForeignKey

class User(BaseModel):

__tablename__ = 'user'

id = Column(Integer, primary_key=True)

age = Column(Integer)

class Friendship(BaseModel):

__tablename__ = 'friendship'

id = Column(Integer, primary_key=True)

user_id1 = Column(Integer, ForeignKey('user.id'))

user_id2 = Column(Integer, ForeignKey('user.id'))

for i in xrange(100):

session.add(User(age=randint(1, 100)))

session.flush() # 或 session.commit(),执行完后,user 对象的 id 属性才可以访问(因为 id 是自增的)

for i in xrange(100):

session.add(Friendship(user_id1=randint(1, 100), user_id2=randint(1, 100)))

session.commit()

session.query(User).filter(User.age <50).delete()

执行这段代码时,你应该会遇到一个错误:

复制代码代码如下:

sqlalchemy.exc.IntegrityError: (IntegrityError) (1451, 'Cannot delete or update a parent row: a foreign key constraint fails (`ooxx`.`friendship`, CONSTRAINT `friendship_ibfk_1` FOREIGN KEY (`user_id1`) REFERENCES `user` (`id`))') 'DELETE FROM user WHERE user.age <%s' (50,)

原因是删除 user 表的数据,可能会导致 friendship 的外键不指向一个真实存在的记录。在默认情况下,MySQL 会拒绝这种 *** 作,也就是 RESTRICT。InnoDB 还允许指定 ON DELETE 为 CASCADE 和 SET NULL,前者会删除 friendship 中无效的记录,后者会将这些记录的外键设为 NULL。

除了删除,还有可能更改主键,这也会导致 friendship 的外键失效。于是相应的就有 ON UPDATE 了。其中 CASCADE 变成了更新相应的外键,而不是删除。

而在 SQLAlchemy 中是这样处理的:

复制代码代码如下:

class Friendship(BaseModel):

__tablename__ = 'friendship'

id = Column(Integer, primary_key=True)

user_id1 = Column(Integer, ForeignKey('user.id', ondelete='CASCADE', onupdate='CASCADE'))

user_id2 = Column(Integer, ForeignKey('user.id', ondelete='CASCADE', onupdate='CASCADE'))

如何连接表?

复制代码代码如下:

from sqlalchemy import distinct

from sqlalchemy.orm import aliased

Friend = aliased(User, name='Friend')

print session.query(User.id).join(Friendship, User.id == Friendship.user_id1).all() # 所有有朋友的用户

print session.query(distinct(User.id)).join(Friendship, User.id == Friendship.user_id1).all() # 所有有朋友的用户(去掉重复的)

print session.query(User.id).join(Friendship, User.id == Friendship.user_id1).distinct().all() # 同上

print session.query(Friendship.user_id2).join(User, User.id == Friendship.user_id1).order_by(Friendship.user_id2).distinct().all() # 所有被别人当成朋友的用户

print session.query(Friendship.user_id2).select_from(User).join(Friendship, User.id == Friendship.user_id1).order_by(Friendship.user_id2).distinct().all() # 同上,join 的方向相反,但因为不是 STRAIGHT_JOIN,所以 MySQL 可以自己选择顺序

print session.query(User.id, Friendship.user_id2).join(Friendship, User.id == Friendship.user_id1).all() # 用户及其朋友

print session.query(User.id, Friendship.user_id2).join(Friendship, User.id == Friendship.user_id1).filter(User.id <10).all() # id 小于 10 的用户及其朋友

print session.query(User.id, Friend.id).join(Friendship, User.id == Friendship.user_id1).join(Friend, Friend.id == Friendship.user_id2).all() # 两次 join,由于使用到相同的表,因此需要别名

print session.query(User.id, Friendship.user_id2).outerjoin(Friendship, User.id

在 Python 语言环境下我们这样连接数据库。

In [1]: from mysql import connector

In [2]: cnx = connector.connect(host="172.16.192.100",port=3306,user="appuser",password="xxxxxx")

但是连接数据库的背后发生了什么呢?

答案

当我们通过驱动程序(mysql-connector-python,pymysql)连接 MySQL 服务端的时候,就是把连接参数传递给驱动程序,驱动程序再根据参数会发起到 MySQL 服务端的 TCP 连接。当 TCP 连接建立之后驱动程序与服务端之间会按特定的格式和次序交换数据包,数据包的格式和发送次序由 MySQL 协议 规定。MySQL 协议:https://dev.mysql.com/doc/internals/en/client-server-protocol.html整个连接的过程中 MySQL 服务端与驱动程序之间,按如下的次序发送了这些包。

MySQL 服务端向客户端发送一个握手包,包里记录了 MySQL-Server 的版本,默认的授权插件,密码盐值(auth-data)。

2. MySQL 客户端发出 ssl 连接请求包(如果有必要的话)。

3. MySQL 客户端发出握手包的响应包,这个包时记录了用户名,密码加密后的串,客户端属性,等等其它信息。

4. MySQL 服务端发出响应包,这个包里记录了登录是否成功,如果没有成功也会给出错误信息。


欢迎分享,转载请注明来源:内存溢出

原文地址: http://outofmemory.cn/sjk/9410471.html

(0)
打赏 微信扫一扫 微信扫一扫 支付宝扫一扫 支付宝扫一扫
上一篇 2023-04-28
下一篇 2023-04-28

发表评论

登录后才能评论

评论列表(0条)

保存