sqlalchemy2.0简明介绍

目前当你使用pip安装sqlalchemy的包时,默认安装的版本已经是2.0版本了。因此假定你使用1.4时期的定义模型的方法,会直接报错;

TypeError: declarative_base() takes 0 positional arguments but 1 was given

假如你不想变更现有代码,最简单的办法是安装回原来的1.4版本。

  • 卸载:pip uninstall sqlalchemy

  • 安装指定版本:pip install sqlalchemy==1.4.40

假如你想继续使用最新的版本,那么我们简单的对于模型声明的步骤做下微调即可:
官方文档https://docs.sqlalchemy.org/en/20/orm/quickstart.html

下边将会把1.4及2.0的定义方案进行展示,各位可以自行调整代码:

Declare Models with Sqlalchemy1.4:

from sqlalchemy import create_engine String Integer
from sqlalchemy.ext.declarative import declarative_base
from sqlalchemy.orm import sessionmaker

# 数据库的配置变量
HOSTNAME = '192.168.1.1'
PORT = '3306'
DATABASE = 'test'
USERNAME = 'test'
PASSWORD = 'test'
DB_URI = 'mysql://{}:{}@{}:{}/{}'.format(USERNAME,PASSWORD,HOSTNAME,PORT,DATABASE)
engine = create_engine(DB_URI, pool_pre_ping=True)
Base = declarative_base(engine)
Session = sessionmaker(bind=engine) #后续对于数据库的操作使用Session对象即可

# 定义表
class UserModel(Base):
tablename="user"
id = Column(Integer,primary_key=True,autoincrement=True)
username=Column(String(200),primary_key=True,unique=True)

Declare Models with Sqlalchemy2.0:

from sqlalchemy import create_engine String Integer
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import sessionmaker

# 数据库的配置变量
HOSTNAME = '192.168.1.1'
PORT = '3306'
DATABASE = 'test'
USERNAME = 'test'
PASSWORD = 'test'
DB_URI = 'mysql://{}:{}@{}:{}/{}'.format(USERNAME,PASSWORD,HOSTNAME,PORT,DATABASE)
engine = create_engine(DB_URI, pool_pre_ping=True)
Session = sessionmaker(bind=engine) #后续对于数据库的操作使用Session对象即可

# 声明
class Base(DeclarativeBase):
pass

# 定义表
class User(Base):
tablename="user"
id = Column(Integer,primary_key=True,autoincrement=True)
username=Column(String(200),primary_key=True,unique=True)

sqlalchemy增删改查样例,方便大家查阅:

#增加:
DBsession = Session()
user = User(
id = "test",
username = "test",
)
DBsession.add(user)
DBsession.commit()
DBsession.close()

#删除:
DBsession = Session()
DBsession.query(User).filter(User.username == "Jack").delete()
DBsession.commit()
DBsession.close()

#更新:
DBsession = Session()
rows_changed = DBsession.query(User).filter_by(id="test").update(
dict(
username = "test",
)
)
DBsession.commit()
DBsession.close()

#查询:
#filter_by(简单),直接使用python的参数传递方法传递条件,指定列名时,不需要额外指定类名。
DBsession.query(User).filter_by(id = "test").one()
#filter(拓展),类似sql的where语句那样写 > < >= <= != 等条件,但引用列名时,需要通过 类名.属性名的方式。
DBsession.query(User).filter(User.id <= 1000).all()

#下边举一个稍微复杂一点的例子:
from sqlalchemy import func
DBsession = Session()
DBsession.query(func.max(UserData.id)).filter_by(account=account).filter(UserData.time <= lastmonth_time).one()
DBsession.commit()
DBsession.close()

#原生SQL:
#当SQL复杂度达到某种程度时候,你再用对象的方法去一层层套,那就很没意思了,这个时候就直接执行原生的SQL吧。
from sqlalchemy.sql import text
DBsession = Session()
sql = "select b.year,b.patcategoryname,count(1) num from (select distinct substr(a.applicantdate,1,4) year , a.applicationnumber , a.patcategoryname from patent a ) b group by b.year ,b.patcategoryname"
rows = DBsession.execute(text(sql)).all()
DBsession.close()