sqlalchemy(一)基本操做

sqlalchemy採用簡單的Python語言，爲高效和高性能的數據庫訪問設計，實現了完整的企業級持久模型。

安裝

1. 須要安裝MySQLdb
2. pip install sqlalchemy

安裝完成後，執行

>>>import sqlalchemy >>>sqlalchemy.__version__

鏈接數據庫

在sqlalchemy中，session用於建立程序與數據庫之間的會話。全部對象的載入和保存都須要經過session對象。

from sqlalchemy import create_engine from sqlalchemy.orm import sessionmaker engine = create_engine('mysql://user:passwd@ip:port/db', echo=True) Session = sessionmaker(bind=engine) session = Session() session.execute('show databases')

其中，echo爲True表明打開logging。

建立一個映射

一個映射對應着一個Python類，用來表示一個表的結構。下面建立一個person表，包括id和name兩個字段。

from sqlalchemy.ext.declarative import declarative_base Base = declarative_base() class Person(Base): __tablename__ = 'person' id = Column(Integer, primary_key=True) name = Column(String(32)) def __repr__(self): return "<Person(name='%s')>" % self.name

添加數據

#建立一個person對象 person = Person(name='jack') #添加person對象，可是仍然沒有commit到數據庫 session.add(person) #commit操做 session.commit()

如何獲取id的？

>>> person = Person(name='ilis') >>> person.id #此時尚未commit到mysql,所以無id >>> session.add(person) >>> person.id #同上 >>> session.commit() 2015-08-18 23:08:23,530 INFO sqlalchemy.engine.base.Engine INSERT INTO person (name) VALUES (%s) 2015-08-18 23:08:23,531 INFO sqlalchemy.engine.base.Engine ('ilis',) 2015-08-18 23:08:23,532 INFO sqlalchemy.engine.base.Engine COMMIT >>> person.id #commit後，能夠獲取該對象的id 2015-08-18 23:08:27,556 INFO sqlalchemy.engine.base.Engine BEGIN (implicit) 2015-08-18 23:08:27,557 INFO sqlalchemy.engine.base.Engine SELECT person.id AS person_id, person.name AS person_name FROM person WHERE person.id = %s 2015-08-18 23:08:27,557 INFO sqlalchemy.engine.base.Engine (5L,) 5L >>>

添加多個數據

session.add_all([
    Person(name='jack'), Person(name='mike') ]) session.commit()

回滾

>>> person = Person(name='test') >>> session.add(person) >>> session.query(person).filter(name=='test') >>> session.query(Person).filter(Person.name=='test').all() 2015-08-18 23:13:23,265 INFO sqlalchemy.engine.base.Engine INSERT INTO person (name) VALUES (%s) 2015-08-18 23:13:23,265 INFO sqlalchemy.engine.base.Engine ('test',) 2015-08-18 23:13:23,267 INFO sqlalchemy.engine.base.Engine SELECT person.id AS person_id, person.name AS person_name FROM person WHERE person.name = %s 2015-08-18 23:13:23,267 INFO sqlalchemy.engine.base.Engine ('test',) [<demo.Person object at 0x7f4e37730510>] >>> session.rollback() 2015-08-18 23:13:37,496 INFO sqlalchemy.engine.base.Engine ROLLBACK >>> session.query(Person).filter(Person.name=='test').all() 2015-08-18 23:13:38,690 INFO sqlalchemy.engine.base.Engine BEGIN (implicit) 2015-08-18 23:13:38,691 INFO sqlalchemy.engine.base.Engine SELECT person.id AS person_id, person.name AS person_name FROM person WHERE person.name = %s 2015-08-18 23:13:38,692 INFO sqlalchemy.engine.base.Engine ('test',) [] >>>

數據查詢

使用Session的query()方法。

#獲取全部數據 session.query(Person).all() #獲取某一列數據,相似於django的get,若是返回數據爲多個則報錯 session.query(Person).filter(Person.name=='jack').one() #獲取返回數據的第一行 session.query(Person).first() #過濾數據 session.query(Person.name).filter(Person.id>1).all() #limit session.query(Person).all()[1:3] #order by session.query(Person).ordre_by(-Person.id) #equal/like/in query = session.query(Person) query.filter(Person.id==1).all() query.filter(Person.id!=1).all() query.filter(Person.name.like('%ac%')).all() query.filter(Person.id.in_([1,2,3])).all() query.filter(~Person.id.in_([1,2,3])).all() query.filter(Person.name==None).all() #and or from sqlalchemy import and_ query.filter(and_(Person.id==1, Person.name=='jack')).all() query.filter(Person.id==1, Person.name=='jack').all() query.filter(Person.id==1).filter(Person.name=='jack').all() from sqlalchemy import or_ query.filter(or_(Person.id==1, Person.id==2)).all()

使用text

from sqlalchemy import text query.filter(text("id>1")).all() query.filter(Person.id>1).all() #同上 query.filter(text("id>:id")).params(id=1).all() #使用:，params來傳參 query.from_statement( text("select * from person where name=:name")).\ params(name='jack').all()

計數

Query使用count()函數來實現查詢計數。

query.filter(Person.id>1).count()

group by的用法

from sqlalchemy import func session.query(func.count(Person.name), Person.name),group_by(Person.name).all()

實現count(*)來查詢表內行數

session.query(func.count('*')).select_from(Person).scalar() session.query(func.count(Person.id)).scalar()