flask-sqlalchemy

flask sqlalchemy

add

user_dict = {
    'username': 'strglee',
    'sex': 'man',
    'birthday': '1993',
}
user = UserModel(**user_dict)                                     
db.session.add(user)                                             
db.session.commit()

update

user = UserModel.query.filter(UserModel.phonenum == '123456').one()
if user:
    user.username = 'strglee'
    db.session.commit()
UserModel.query.filter(UserModel.phonenum == '123456').update({'username': 'strglee'})
session.query(JobModel).filter(JobModel.id.in_([1, 3, 5, 7, 9])).update(dict(local_job_id = 0,local_company_id=0),synchronize_session=False)

delete

UserModel.query.filter(UserModel.phonenum == '123456').delete()

select

UserModel.query.filter(UserModel.username=='strglee').one()
UserModel.query.filter_by(username='strglee').one()
UserModel.query.filter_by(username='strglee').first()
UserModel.query.filter_by(username='strglee').first_or_404()

join 查询

UserModel.query.join(AddressModel,AddressModel.user_id == UserModel.id).\    
    add_columns(
            UserModel.username,                         
            AddressModel.city,                             
            AddressModel.address,                               
            UserModel.sex).\
    limit(10).all()

paginate

paginate = UserModel.query.order_by(SeoModel.dt_update.desc()).paginate(page, 15, False)    
query4.filter(not_(User.name == None)).all() # not  
query4.filter(User.name != None).all()  

func

from sqlalchemy import func
rows = session.query(Person).count()
c.count = Session.query(func.count(Person.id)).scalar()
c.avg = Session.query(func.avg(Person.id).label('average')).scalar()
c.sum = Session.query(func.sum(Person.id).label('average')).scalar()
c.max = Session.query(func.max(Person.id).label('average')).scalar() 

group by

c.coutg = Session.query(func.count(Person.id).label('count'), Person.name ).group_by(Person.name).all()

destinct

from sqlalchemy import distinct
# count distinct "name" values
session.query(func.count(distinct(User.name)))

sqlalchemy

#!/usr/bin/env python
#-*- coding: utf-8 -*-
from sqlalchemy import create_engine, Column, Integer, String, func
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker
Base = declarative_base()
class StudentInfo(Base):
    __tablename__ = 'stuinfo'
    id = Column(Integer, primary_key=True)
    name = Column(String)
    cls = Column(String)
    cert_count = Column(Integer)
    def __init__(self, name, cls, cert_count):
        self.name = name
        self.cls = cls
        self.cert_count = cert_count
engine = create_engine('sqlite:///:memory:', echo=True)
Base.metadata.create_all(engine)
Session = sessionmaker(engine)
session = Session()
data = [
    [u'学生1', u'A2', 1],
    [u'学生1', u'A1', 4],
    [u'学生1', u'A2', 4],
    [u'学生1', u'A1', 1],
    [u'学生1', u'A1', 5],
    [u'学生1', u'A2', 1]]
reocords = [StudentInfo(*record) for record in data]
session.add_all(reocords)
session.commit()

sqlalchemy way

rs = session.query(StudentInfo.cls, func.sum(StudentInfo.cert_count)) \
        .group_by(StudentInfo.cls).all()
for row in rs:
    print row[0], row[1]
""" Output
A1 10
A2 6
"""

sql way

sql = 'select cls, sum(cert_count) from stuinfo group by cls'
rs = engine.execute(sql) 
for row in rs:
    print row[0], row[1]
""" Output
A1 10
A2 6
"""

group by and order by

jobs = session.query(JobModel.salary_start,func.count(JobModel.id).label('cnt')).filter(JobModel.id.between(1,10000)).group_by(JobModel.salary_start).order_by('cnt desc').all()
jobs = session.query(JobModel.salary_start,func.count(JobModel.id).label('cnt')).filter(JobModel.id.between(1,10000)).group_by(JobModel.salary_start).order_by(desc('cnt')).all()