标签：sqlalchemy

SQLAlchemy查询过滤器

SQLAlchemy查询执行函数

# coding=utf-8
from urllib import quote_plus as urlquote
from sqlalchemy import create_engine, Table, Column, Integer, String, MetaData, ForeignKey
import MySQLdb

创建数据库连接

用此方法链接可以指定UTF8编码，同时解决密码中含有特定字符，比如含有@，则把密码部分进行URL编码
echo=True 会显示每条执行的 SQL 语句
max_overflow 最大链接数

connstr = "mysql+mysqldb://root:%[email protected]:3306/dbname?charset=utf8" % urlquote('password')
engine = create_engine(connstr,echo=True,max_overflow=5)

获取元数据

metadata = MetaData()

定义表

user = Table('user', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(20)),
    )

color = Table('color', metadata,
    Column('id', Integer, primary_key=True),
    Column('name', String(20)),
    )

创建数据表，如果数据表存在，则忽视

metadata.create_all(engine)

创建链接

conn = engine.connect()

增删改查

#插入数据
engine.execute(
    "INSERT INTO color( name) VALUES ('test');"
)
result = engine.execute('select * from color')


print(result.fetchall())

conn.execute(user.insert(),{'name':'test'})

sql = user.insert().values(name='test2')
conn.execute(sql)

#删除一条user表里的 条件是id大于1的
#sql = user.delete().where(user.c.id > 1)
#执行
#conn.execute(sql)

#更新
#把名字为test的修改为aaa
#sql = user.update().where(user.c.name == 'test').values(name='aaa')
#conn.execute(sql)

print "查询user表里的内容"
sql = select([user, ])
res =conn.execute(sql)
print res.fetchall()

print '查询user表下的id'
sql = select([user.c.id, ])
res =conn.execute(sql)
print res.fetchall()

print '查询user表和color表的name，条件是user表的id1=color的id1'
sql = select([user.c.name, color.c.name]).where(user.c.id==color.c.id)
res =conn.execute(sql)
print res.fetchall()

print '查询user表的name，并按照条件排序'
#按照名字排序
sql = select([user.c.name]).order_by(user.c.name)
res =conn.execute(sql)
print res.fetchall()

print '按照id排序'
sql = select([user.c.name]).order_by(user.c.id)
res =conn.execute(sql)
print res.fetchall()

print '查询user表的name，并按照条件分组'
sql = select([user]).group_by(user.c.name)
res =conn.execute(sql)
print res.fetchall()

关闭链接

#关闭链接
conn.close()

user_modules.py

from datetime import datetime
from sqlalchemy import Column,Integer,String,Boolean,DateTime,ForeignKey
from sqlalchemy.orm import  relationship
from .connect import Base,session

class User(Base):
    __tablename__='user'
    id=Column(Integer,primary_key=True,autoincrement=True)
    username=Column(String(20),nullable=False)
    passwd=Column(String(50),nullable=False)
    createtime=Column(DateTime,default=datetime.now)
    _locked=Column(Boolean,default=False,nullable=False)
    #在modules中写好查询条件，使用时直接调用
    @classmethod
    def all(cls):
        return session.query(cls).all()
    @classmethod
    def by_name(cls,username):
        return session.query(cls).filter_by(username=username).all()
    @property
    def locked(self):
        return self._locked

    def __repr__(self):
        return '<User(id=%s,username=%s,passwd=%s,createtime=%s,_locked=%s)>'%(
          self.id,
          self.username,
          self.passwd,
          self.createtime,
          self._locked
        )
class UserDetails(Base):
     __tablename__='user_details'
     id=Column(Integer,primary_key=True,autoincrement=True)
     id_card=Column(Integer,nullable=True,unique=True)
     last_login=Column(DateTime)
     login_num=Column(Integer,default=0)
     user_id=Column(Integer,ForeignKey('user.id'))
#bakcref建立反向索引，
     userdetails_for_foreignkey=relationship('User',backref='details',uselist=False,cascade='all')
     def __self__(self):
         return '<UserDetails(id=%s,id_card=%s,last_login=%s,login_num=%s,user_id=%s)>'%(
             self.id,
             self.id_card,
             self.last_login,
             self.login_num,
             self.user_id
         )
if __name__=='__main__':
    Base.metadata.create_all()

query.py

from data.user_modules import User,session,UserDetails
#带条件查询
raw=session.query(User).filter_by(username='nanian').all()
raw=session.query(User).filter_by(username='nanian') #去掉.all()原生sql
raw=session.query(User).filter(User.username  =='nanian').all()
raw=session.query(User.username).filter(User.username  !='nanian').all()
raw=session.query(User.username).filter(User.username  !='nanian').first()
raw=session.query(User.username).filter(User.username  !='nanian').one() #如果前面查出的是多条数据则报错
print(session.query(User).get(2)) #根据主键查，会自己找主键
print(raw)

#限制查询结果数
print(session.query(User).filter(User.username!='nanian').limit(3).all())#前三行
print(session.query(User).filter(User.username!='nanian').offset(3).all())#第三行以后
print(session.query(User).filter(User.username!='nanian').slice(1,3).all())#2,3行


#排序
from sqlalchemy import desc
raw=session.query(User).filter(User.username  !='nanian').order_by(User.username).all()
raw=session.query(User).filter(User.username  !='nanian').order_by(desc(User.username).all()#逆序

#模糊查询 尽量少用模糊查询，效率低
from sqlalchemy import or_
raw=session.query(User).filter(User.username!='nanian').all()
raw=session.query(User).filter(User.username.like('n%').all()
raw=session.query(User).filter(User.username.notlike('n%').all()
raw=session.query(User).filter(User.username.in_(['nanian','a']).all()) #加下划线表示和python关键字作区分
raw=session.query(User).filter(User.username.isnot(None),User.passwd=='123').all()) #多条件
raw=session.query(User).filter(or_(User.username.isnot(None),User.passwd=='123')).all()) #或

raw=session.query(User).filter(User.username==None).all())

#聚合函数
from sqlalchemy import func,extract
print(session.query(User.passwd,func.count(User.id)).group_by(User.passwd).all())
print(session.query(User.passwd,func.count(User.id)).group_by(User.passwd).
      having(func.count(User.id)>1) all())
print( session.query(extract('minute',User.createtime).label('minute'),
                     func.count(User.id)).group_by('minute').all() )  #提取分钟，按分钟分组

#多表查询
raw=session.query(User,UserDetails).all()
raw=session.query(User,UserDetails).filter(UserDetails.id==User.id) all()# cross join
raw=session.query(User.username,UserDetails.last_login).
    join(UserDetails, UserDetails.id==User.id) all()# inner join
raw=session.query(User.username,UserDetails.last_login).
    outerjoin(UserDetails, UserDetails.id==User.id) all()
# outer join代表left join 左连接，右连接将表反过来(sqlalchemy没有rightjoin),小表左连接右表效率高

q1=session.query(User.id)
q2=session.query(UserDetails.id)
raw=q1.union(q2).all()

from sqlalchemy import  all_,any_
sql_0=session.query(UserDetails.last_login).subquery() #声明子表
raw=session.query(User).filter(User.createtime >all_(sql_0)).all()
raw=session.query(User).filter(User.createtime >any_(sql_0)).all()

#原生sql
sql_1='''
select * from `user`
'''
raw=session.execute(sql_1)
#print(raw,dir(raw))
#print(raw.fetchone())
#print(raw.fetchmany())
#print(raw.fetchall())

for i in raw:
    print(i)