本文介绍sqlalchemy在python中的使用,不涉及到flask或者tornado框架,需要的时候融入到框架项目中即可。

一、依赖包的安装

• 1、安装sqlalchemy

  pip3 install sqlalchemy
  

• 2、安装pymysql

  pip3 install pymysql
  

二、创建连接(使用python代码连接到mysql)

• 1、新建数据库

  mysql> create database sqlalchemy_data charset=utf8;
  

• 2、新建一个文件connect.py

• 3、导包及数据库的基本信息

  from sqlalchemy import create_engine
  
  HOSTNAME = '127.0.0.1'
  PORT = '3306'
  DATABASE = 'sqlalchemy_data'
  USERNAME = 'root'
  PASSWORD = 'root'
  

• 4、使用pymysql创建连接的db_url

  db_url = 'mysql+pymysql://{username}:{password}@{hostname}:{port}/{database}?charset=utf8'.format(
      username=USERNAME,
      password=PASSWORD,
      hostname=HOSTNAME,
      port=PORT,
      database=DATABASE
  )
  

• 5、创建一个引擎

  # 创建引擎
  engine = create_engine(db_url)
  

• 6、测试连接是否成功

  if __name__ == "__main__":
      connection = engine.connect()
      result = connection.execute('select 1')
      print(result.fetchone())
  

• 7、完整代码

  from sqlalchemy import create_engine
  
  HOSTNAME = '127.0.0.1'
  PORT = '3306'
  DATABASE = 'sqlalchemy_data'
  USERNAME = 'root'
  PASSWORD = 'jianshuihen128'
  
  db_url = 'mysql+pymysql://{username}:{password}@{hostname}:{port}/{database}?charset=utf8'.format(
      username=USERNAME,
      password=PASSWORD,
      hostname=HOSTNAME,
      port=PORT,
      database=DATABASE
  )
  
  # 创建引擎
  engine = create_engine(db_url)
  
  if __name__ == "__main__":
      connection = engine.connect()
      result = connection.execute('select 1')
      print(result.fetchone())
  

三、使用sqlalchemy创建数据表

• 1、创建一个modules的文件,存放数据模型

• 2、在connect.py文件中添加连接的session与declarative_base

  from sqlalchemy.ext.declarative import declarative_base
  from sqlalchemy.orm import sessionmaker
  
  # sessionmaker生成一个session类
  Session = sessionmaker(bind=engine)
  dbSession = Session()
  
  Base = declarative_base(engine)
  

• 3、在modules包下创建一个user_module.py的文件

  import datetime
  from uuid import uuid4
  from sqlalchemy import Column, Integer, String, DateTime, Boolean
  from sqlalchemy_demo.connect import Base
  
  class UserModule(Base):
      """
      创建一个用户的数据模型
      """
      __tablename__ = 'user'
  
      uuid = Column(String(36), unique=True, nullable=False, default=lambda: str(uuid4()), comment='uuid')
      id = Column(Integer, primary_key=True, autoincrement=True, comment='用户id')
      user_name = Column(String(30), nullable=False, unique=True, comment='用户名')
      password = Column(String(64), nullable=False, comment='用户密码')
      createtime = Column(DateTime, default=datetime.datetime.now, comment='创建时间')
      updatetime = Column(DateTime, default=datetime.datetime.now, comment='修改时间')
      is_lock = Column(Boolean, default=False, nullable=False, comment='是否锁住用户')
      
      def __repr__(self):
          return 'User(uuid={uuid}, id={id}, user_name={user_name}, password={password}, createtime={createtime}, updatetime={updatetime}, is_lock={is_lock})'.format(
              uuid=self.uuid, id=self.id, user_name=self.user_name, password=self.password, createtime=self.createtime,
              updatetime=self.updatetime, is_lock=self.is_lock)
  

• 4、运动代码创建数据表(user_module.py文件下)

  if __name__ == "__main__":
      Base.metadata.create_all()
  

• 5、查看数据表信息

  mysql> desc user;
  +------------+-------------+------+-----+---------+----------------+
  | Field      | Type        | Null | Key | Default | Extra          |
  +------------+-------------+------+-----+---------+----------------+
  | uuid       | varchar(36) | NO   | UNI | NULL    |                |
  | id         | int(11)     | NO   | PRI | NULL    | auto_increment |
  | user_name  | varchar(30) | NO   | UNI | NULL    |                |
  | password   | varchar(64) | NO   |     | NULL    |                |
  | createtime | datetime    | YES  |     | NULL    |                |
  | updatetime | datetime    | YES  |     | NULL    |                |
  | is_lock    | tinyint(1)  | NO   |     | NULL    |                |
  +------------+-------------+------+-----+---------+----------------+
  7 rows in set (0.01 sec)
  
  mysql> 
  

四、对数据表简单的增删改查

• 1、单独创建一个test_user.py的测试文件

• 2、确保前面在connect.py文件中新增的

  from sqlalchemy.orm import sessionmaker
  
  # sessionmaker生成一个session类
  Session = sessionmaker(bind=engine)
  dbSession = Session()
  

• 3、新增单条数据

  from sqlalchemy_demo.connect import dbSession
  from sqlalchemy_demo.modules.user_module import UserModule
  
  def add_user():
      row = UserModule(user_name='张三', password='123')
      dbSession.add(row)
      dbSession.commit()
  
  if __name__ == "__main__":
      add_user()
  

• 4、同时新增多条数据

  from sqlalchemy_demo.connect import dbSession
  from sqlalchemy_demo.modules.user_module import UserModule
  
  def add_user():
      dbSession.add_all([
          UserModule(user_name='王五', password='123'),
          UserModule(user_name='马六', password='123'),
          UserModule(user_name='赵五', password='123'),
      ])
      dbSession.commit()
  
  if __name__ == "__main__":
      add_user()
  

• 5、查询数据

  ...
  def query_user():
      rows = dbSession.query(UserModule).all()
      print(rows)
  ...
  

• 6、修改数据

  ...
  def update_user():
      row = dbSession.query(UserModule).filter(UserModule.id == 1).update({UserModule.password: '234'})
      print(row)
      dbSession.commit()
  ...
  

• 7、删除数据

  ...
  def delete_user():
      row = dbSession.query(UserModule).filter(UserModule.id == 3)[0]
      print(row)
      dbSession.delete(row)
      dbSession.commit()
  ...