Python调用sqlAlchemy

import logging
from sqlalchemy import Column, String, create_engine
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 创建对象的基类
Base = declarative_base()


# 创建User对象
class User(Base):
    # 使用InnoDB，UTF-8编码
    __table_args__ = {
        'mysql_engine': 'InnoDB',
        'mysql_charset': 'utf8'
    }
    # 表的名字
    __tablename__ = 'User'
    # 表结构
    id = Column(String(20), primary_key=True)
    name = Column(String(20))


def make_connect():
    # 初始化数据库连接
    engine = create_engine("mysql+pymysql://root:123@localhost:3306/test", echo=True)
    # 创建所有Base子类表
    Base.metadata.create_all(engine)
    # 创建DBSession类型
    DBSession = sessionmaker(bind=engine)
    # 创建session
    session = DBSession()
    return session

“mysql+pymysql”使用 MySQL-Pymysql 来连接，“root”和“123”分别是用户名和密码，“localhost”是数据库的域名，“test”是使用的数据库名，“charset”指定了连接时使用的字符集。

添加数据：

  new_user = User(id='4', name='root')
    session = make_connect()
    session.add(new_user)
    try:
        session.commit()
    except BaseException as e:
        logging.exception(e)
    finally:
        session.close()

添加结果：

 删除数据需要先查询数据再删除：

 session = make_connect()
    user = session.query(User).filter(User.id == '4').first()
3     session.delete(user)
    try:
        session.commit()
    except BaseException as e:
        logging.exception(e)
    finally:
        session.close()

删除结果：

 

 更新数据(依然先查询数据持久化后修改)：

session = make_connect()
    session.query(User).filter(User.id == '4').first().id = '1'
    try:
        session.commit()
    except BaseException as e:
        logging.exception(e)
    finally:
        session.close()

以上将id=4 的数据查询出后id更新为1