sqlalchemy 简介

#! /usr/bin/env python3

# -*- coding:utf-8 -*-

 

#use SQLAlchemy

 ORM

#ORM:Object-Relational Mapping ,把关系数据库的结构映射到对象上，SQLAlchemy是Python中有名的ORM框架

 

#导入sqlalchemy 

from sqlalchemy import Column,String,create_engine

from sqlalchemy.orm import sessionmaker

from sqlalchemy.ext.declarative import declarative_base

 

 

#创建对象的基类：

Base=declarative_base()  #这个语句创建了一个Base类，这个类的子类可以自动与一个表关联

 

#定义创建表的函数

def init_db():

    Base.metadata.create_all(engine) #该语句会找到base的所有子类，并在数据库中创建这些表，drop_all则是删除这些表

#定义删除表的函数

def drop_db():

    Base.metadata.drop_all(engine)

 

#定义User表对象：

class User2(Base):

    #表名：

    __tablename__='user2'

    #表结构：

    id=Column(String(20),primary_key=True)

    name=Column(String(20))

    #初始化数据库连接：

DB_CONNECT_STRING='mysql+pymysql://rduser:1qaz@WSX@172.18.28.110:3306/test?charset=utf8'

#'mysql+pymysql'指定了使用mysql-python来连接  数据库名test和连接时使用的字符集charset=utf8可以省略。

engine=create_engine(DB_CONNECT_STRING,echo=True)

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

 

#创建DBSession类型：

DBSession=sessionmaker(bind=engine)

 

#sessionmaker()会生成一个数据库会话类。这个类的实例可以当成一个数据库连接，它同时还记录了一些查询的数据，并决定什么时候执行sql语句。

 

session=DBSession()

 

#init_db()

#以上代码完成SQLAlchemy 的初始化和具体每个表的class定义。如果有多个表就继续定义其他class,例School:

'''

class School(Base):

    __tablename__='school'

    id=...

    name=...

'''

#create_engine()用来初始化数据库连接。SQLAlchemy用一个字符串表示连接信息：

# ‘数据库类型+数据库驱动名称：//用户名：口令@机器地址：端口号/数据库名’

#根据需要替换掉用户名、口令等信息即可

 

#由于有了ORM，我们向数据库表中添加一行记录，可以视为添加一个User对象：

 

#创建session对象：

session=DBSession()

#创建新User对象,添加新记录：

new_user=User2(id='1',name='Glace')

#添加到session:

session.add(new_user)  #写入数据

user=User2(id='3',name='Smith')

session.add(user)

#提交即保存到数据库：

session.commit()

#关闭session:

session.close()

 

#可见，关键是获取session,然后把对象添加到session，最后提交并关闭。DBSession对象可视为当前数据库连接。

 

#有了ORM,查询出来的可以不再是tuple,而是User对象，sqlalchemy提供的查询接口如下：

#创建Session:

session=DBSession()

#创建DBSession类型：Query查询，filter是where条件，最后调用one()返回唯一行，如果调用all()则返回所有行：

print(session.query(User2).filter(User2.id=='5').one())

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

#打印类型和对象的name属性：

print('type:',type(User2))

print('name:',User2.name)

#关闭Session:

session.close()

#type: <class 'sqlalchemy.ext.declarative.api.DeclarativeMeta'>

#name: User2.name

 

#可见，ORM就是把数据库表的行与相应的对象建立关联，互相转换。

#由于关系数据库的多个表还可以用外键实现一对多、多对多等关联，相应的，ORM矿建也可以提供两个对象之间的一对多、多对多等功能。

 

#例如，如果一个User拥有多个Book,就可以定义一对多的关系：

'''

class User(Base):

    __tablename__ = 'user'

 

    id = Column(String(20), primary_key=True)

    name = Column(String(20))

    # 一对多:

    books = relationship('Book')

 

class Book(Base):

    __tablename__ = 'book'

 

    id = Column(String(20), primary_key=True)

    name = Column(String(20))

    # “多”的一方的book表是通过外键关联到user表的:

    user_id = Column(String(20), ForeignKey('user.id'))

'''

#当我们查询一个User对象时，该对象的books属性将返回一个包含若干个Book对象的list

#ORM框架的作用就是把数据库表的一行记录与一个对象互相做自动转换。

 ---------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------------

 

pooling

http://docs.sqlalchemy.org/en/devel/

sqlalchemy通过被用作ORM框架，也可以当作普通的数据库接口使用，内带数据库连接池

from sqlalchemy import create_engine

DB_conf="dialect+driver://username:password@host:port/database"   

#dialect表示数据库的名字（比如mysql,oracle,postgresql等），diver表式连接数据库使用的DBAPI模块名（比如psycopg2,pyodbc,cx_oracle等）

example="mysql+pymysql://user:mima@192.168.2.251:3306/mydatabase?charset=utf8"                       mysql

postgresql_DBConf='postgresql+psycopg2://scott:tiger@localhost/mydatabase'                                  postgresql

oracle_DBConf='oracle://scott:tiger@127.0.0.1:1521/sidname'                                                             oracle

ms_DBConf='mssql+pymssql://scott:tiger@hostname:port/dbname'                                                      sql server

sqlite_DBConf='sqlite:///foo.db'                                                                                                            sqlite

engine=create_engine(DB_conf,pool_size=6,max_overflow=8,pool_recycle=7*60,echo=True)      #echo是sqlalchemy内部的日志输出开关，默认为false

https://docs.sqlalchemy.org/en/13/core/engines.html

create_engine(*args,**kwargs)

DB_conf   为位置参数必填，其他的都为选填参数 

pool_size=5 ,连接池里保持的连接数

pool=None 参数为一个已创建的pool实例，比如QueuePool实例，如果为None，则直接使用QueuePool创建数据库连接池引擎。

poolclass：创建连接池时使用的pool子类，与pool参数不同的是，这个参数是一个类名，而不是一个类实例。

max_overflow=10   ,表示除了设置的pool_size连接数，还可以额外打开的连接数，仅适用于QueuePool.

encoding='utf8' :  例latin1,主要用于解决python2自己和许多DBAPI交互时的编码问题，在python3中几乎不会用到这个参数

echo=False ,  sqlalchemy内部的日志输出开关，默认关闭，不打印sqlalchemy的内部日志

isolation_level  数据库隔离等级，可选参数为："SERIALIZABLE", "REPEATABLE_READ", "READ_COMMITTED", "READ_UNCOMMITTED" and "AUTOCOMMIT"。会因每个后端与数据库类型不同而不同。https://docs.sqlalchemy.org/en/13/dialects/mysql.html#mysql-isolation-level

label_length=None 整型参数，用于限制表格中自增列的id过大，如果小于6位数，则以实际"_(counter)”表示。如果为None，将会取dialect.max_identifier_length的值（这个值可能会受参数create_engine.max_identifier_length的影响）create_engine.label_length的值不会大于create_engine. max_idenfier_length。

listeners 为列表参数，列表元素为一个或多个可接收连接池事件的PoolListener对象

engine是一个数据库连接管理器, 创建engine时，实际上就创建了一个连接池pool和会话语法dialect

http://docs.sqlalchemy.org/en/devel/core/engines.html，sqlalchemy自带了很多数据连接引擎，个别需要安装相应模块，详见Dialectshttp://docs.sqlalchemy.org/en/devel/dialects/index.html

通过create_engine被创建时并没有实际连接数据库（表现为惰性连接），只有当engine。execute()或engine。connect()被首次执行时，才会创建实际的数据库连接。

创建个性化连接

def connect():
    return psycopg.connect(user='scott',host='localhost')

db=create_engine('postgresql://',creator=connect)

 

conn=engine.connect()

results=conn.execute('select username from users')  #ins可以是通过sqlalchemy创建的orm对象，也可以是sql语句

for row in results:
    print(username',row['username'])

conn.close()

这个conn是一个connection实例，为实际DBAPI连接的代理对象

results是代理结果实例，同DBAPI的cursor相联系，拥有大多cursor的特性。

当代理结果中的所有返回的rows被取出后，cursor就自动关闭了，例对于一个update操作不会返回row，cursor就会立刻关闭。

直接使用conn.execute(),是自动提交commit模式.即隐式提交

 

显式创建事务

conn=engine.connect()

trains=conn.begin()

try:

　　results=conn.execute()

　　r1=conn.execute(table.insert(),col1=7,col2='this is some data')

      trans.commit()

except:

    trans.rollback()

　raise

也可写作

with engine.begin() as connection:

    r1=conn.execute(table.insert(),col1=7,col2='this is some data')

　r2=conn.execute()

或

with connection.begin() as trans:
    r1 = connection.execute(table1.select())
    connection.execute(table1.insert(), col1=7, col2='this is some data')

r1.fetchall() 调用完一次后，再次调用就会返回一个空列表

r1.fetchmany()

r1.fetchone()

r1.lastrowid

r1.rowcount执行update或delete时返回的受影响行数

r1.close()关闭cursor 

conn.close()

Pooling连接池

https://docs.sqlalchemy.org/en/latest/core/pooling.html

连接池是一个用于维护数据库 的 长时间连接，高效复用，对并发连接数 进行管理的标准技术

可以在创建engine时就传入参数进行自定义

engine=create_engine('postgresql://me@localhost/mydb',pool_size=20,max_overflow=0)

当engine执行connect()或execute()时，就会向连接池获取一个连接，连接池的默认模式为QueuePool,根据需要创建连接的模式。

QueuePool的默认连接数是5，最大连接数是10. 应该保持一个数据库一个engine,而不是为每次连接创建一个engine

所有的sqlalchemy连接池都不会预先创建连接，直到第一次被使用。

使用不同的pool模式,通过poolclass：

from sqlalchemy.pool importQueuePool 

engine=create_engine('sqlite://file.db',poolclass=QueuePool)

 

禁用连接池，使用NullPool

engine=create_engine('postgresql+pscopg2://scott:tiger@localhost/test',poolclass=NullPool)

所有的pool类接收一个creator参数，用于创建一个新连接。 create_engine()接收这个参数，用来连接数据库

import sqlalchemy.pool as pool 

import psycopg2 

def getconn():

    c=psycopg2.connect(username='ed',host='127.0.0.1',dbname='test')

    return c 

engine=create_engine('postgresql+psycopg2://',creator=getconn)

单独使用pool

mypool=pool.QueuePool(getconn,max_overflow=10,pool_size=5) 

cursor=conn.cursor() 

cursor.execute('select foo')

conn.close()

调用完conn.close()这个连接就被关闭，返还回连接池

 连接池事件Pool Event 

连接池支持一个事件接口用于处理上一个连接，基于每一个新的连接

关闭连接的处理

连接池可以刷新所有的单个连接，和它的连接设置，将池中之前的连接设置为无效。

比较适合用于数据库重启后导致之前池中所有连接失效时连接池的连接回收。有两者方法来实现它：

悲观处理关闭（每有一个新的的连接就发出一个简单的连接测试比例select 1,一旦该次结果返回异常，就认为连接断开了，这个连接会立即被回收，该连接之前的所哟连接也会被标记为无效，都会在使用之前被回收）

engine = create_engine("mysql+pymysql://user:pw@host/db", pool_pre_ping=True)


乐观处理的方式之一  设置连接池的回收时长
e=create_engine("mysql://scott:tiger@localhost/test",pool_recycle=3600)
这样会使所有超过3600秒的连接被设置为无效连接并被替换。标记为无效的操作只在检查的过程的发生，而不是在每一个已被检查过的连接。 
pool_recycle是连接池自带的功能，依赖于一个engine是否在被使用。
这个设置适用于MySQL，MySQL会在间隔一个固定时间后自动关闭连接

 

在多进程中使用连接池

http://docs.sqlalchemy.org/en/devel/core/pooling.html

通常不建议在子进程中单独创建engine.

既不是在子进程中创建新的engine,也不是使用一个已存在的engine,而是调用engine.dispose(),在子进程开始connection之前。该方法会移除连接池中所有已存在的连接，重新创建所有的连接。写作如下：

TCP connections are represented as file descriptors, which usually work across process boundaries, meaning this will cause concurrent access to the file descriptor on behalf of two or more entirely independent Python interpreter states.

这意味这当前的到这个文件的连接代表这两个或者更多的完全独立的python接口状态,(可能是说，这个数据库的连接也代表了其他进程里的连接）

方法之一就是在 子进程创建连接之前调用 engine.dispose() 方法，dispose()是用于清除其他子进程中的连接，使在该进程的连接不影响其他的进程中的连接

engine=create_engine('...')

def run_in_process():
    engine.dispose()

    with eng.connect() as conn: 

　　  conn.execute('...') 

p=Process(tartget=run_in_process)

 另一种方式是设置连接池的事件，让其他的连接在这个子进程里自动失效，这种方式可能更合适

from sqlalchemy import event

from sqlalchemy import exc 

import os 

eng=create_engine('..')

@event.listens_for(engine,"connect")

def connect(dbapi_connection,connection_record): 

　　connection_record.info['pid']=os.getpid() 

@event.listens_for(engine,"checkout")

def checkout(dbapi_connection,connection_record,connection_proxy):

　　pid=os.getpid()

　　if connection_record.info['pid']!=pid:

　　　　connection_record.connection=connection_proxy.connection=None

　　　　raise exc.DisconnectionError(

　　　　　　　"Connection record belongs to pid %s,"

　　　　　　　"attempting to check out in pid %s"%(connection_record.info['pid'],pid))

 

 

 

 

将sqlalchemy内置log部分加入日志,echo=True只是用于控制台的输出。

import logging

logging.basicConfig()

logging.getLogger('sqlalchemy.engine').setLevel(logging.INFO)