python SQLAlchemy

一、连接及创建类

from sqlalchemy import create_engine
from sqlalchemy.orm import session
from sqlalchemy.ext.automap import automap_base
# 连接数据库:
engine = create_engine('mysql+pymysql://root:password@localhost:3306/mydb')
# 创建与表名匹配的映射类:
base = automap_base()
base.prepare(engine, reflect=true)
metadata = base.classes.fn_metadata
resource = base.classes.fn_resource
task = base.classes.fn_task
config = base.classes.sys_config
user = base.classes.sys_user
db_session = session(engine)

 

二、新增

#新增
#方式一
session.add(user(user_id=1, user_name='test'))
session.add(config(config_id=1, config_name='test'))
session.commit()
session.close()
#方式二
session.add_all([
    user(user_id=1, user_name='test'),
    config(config_id=1, config_name='test')
])
session.commit()
session.close()

 

三、查询

#结果的类型
instance
instance of list
keyed tuple of list
value of list

#基本查询
session.query(user).filter_by(username='abc').all()
session.query(user).filter(user.username=='abc').all()
session.query(blog).filter(blog.create >= 0).all()
session.query(blog).filter(blog.create >= 0).first()
session.query(blog).filter(blog.create >= 0 | blog.title == 'a').first()
session.query(blog).filter(blog.create >= 0 & blog.title == 'a').first()
session.query(blog).filter(blog.create >= 0).offset(1).limit(1).scalar()
session.query(user).filter(user.username ==  'abc').scalar()
session.query(user.id).filter(user.username ==  'abc').scalar()
session.query(blog.id).filter(blog.create >= 0).all()
session.query(blog.id).filter(blog.create >= 0).all()[0].id
dict(session.query(blog.id, blog.title).filter(blog.create >= 0).all())
session.query(blog.id, blog.title).filter(blog.create >= 0).first().title
session.query(user.id).order_by('id desc').all()
session.query(user.id).order_by('id').first()
session.query(user.id).order_by(user.id).first()
session.query(user.id).order_by(-user.id).first()
session.query('id', 'username').select_from(user).all()
session.query(user).get('16e19a64d5874c308421e1a835b01c69')

#多表查询
session.query(blog, user).filter(blog.user == user.id).first().user.username
session.query(blog, user.id, user.username).filter(blog.user == user.id).first().id
session.query(blog.id,
              user.id,
              user.username).filter(blog.user == user.id).first().keys()

#条件查询
from sqlalchemy import or_, not_
session.query(user).filter(or_(user.id == '',
                               user.id == '16e19a64d5874c308421e1a835b01c69')).all()
session.query(user).filter(not_(user.id == '16e19a64d5874c308421e1a835b01c69')).all()
session.query(user).filter(user.id.in_(['16e19a64d5874c308421e1a835b01c69'])).all()
session.query(user).filter(user.id.like('16e19a%')).all()
session.query(user).filter(user.id.startswith('16e19a')).all()
dir(user.id)

#函数
from sqlalchemy import func
session.query(func.count('1')).select_from(user).scalar()
session.query(func.count('1'), func.max(user.username)).select_from(user).first()
session.query(func.count('1')).select_from(user).scalar()
session.query(func.md5(user.username)).select_from(user).all()
session.query(func.current_timestamp()).scalar()
session.query(user).count()

 

四、删除

#删除
#方式一
session.query(user).filter(user.user_id == 1).first().delete(）
session.commit()
session.close()
#方式二
user = session.query(user).filter(user.user_id == 1).first()
session.delete(user)
session.commit()
session.close()

 

五、修改

#修改
#方式一
session.query(user).filter(user.user_name == 'abc').update({'name': 'test'})
session.commit()
session.close()
#方式二
user = session.query(user).filter_by(user_name='abc').scalar()
user.name = 'test'
session.commit()
session.close()

 

六、异常

try:
    pass
except exception as e:
    logging.error('[x] %s' % e)
    print('[x] %s' % e)  

#必填项为空
sqlalchemy.exc.integrityerror
#字段类型错误
sqlalchemy.exc.internalerror

 

 

 

引用说明：

http://www.codexiu.cn/python/sqlalchemy%e5%9f%ba%e7%a1%80%e6%95%99%e7%a8%8b/73/529/

 

转载请注明出处。