Python：数据库配置与迁移

Python
专业 编程语言
发布日期:   2024-01-07
更新日期:   2025-05-03
文章字数:   1.9k
阅读时长:   9 分
阅读次数:  

1.使用ORM创建表

from typing import List
from typing import Optional
from sqlalchemy import ForeignKey,String
from sqlalchemy import create_engine
from sqlalchemy.orm import DeclarativeBase
from sqlalchemy.orm import Mapped
from sqlalchemy.orm import mapped_column
from sqlalchemy.orm import relationship
from sqlalchemy.orm import Session
#建立数据引擎
engine = create_engine("mysql+pymysql://remote:qwer0487@192.168.198.186:3306/test", echo=False)
class Base(DeclarativeBase):
    pass
#声明表模型
class User(Base):
    __tablename__ = "user_account"#表名
    
    id: Mapped[int] = mapped_column(primary_key=True)
    name: Mapped[str] = mapped_column(String(30))
    fullname: Mapped[Optional[str]] = mapped_column(String(50))
    addresses: Mapped[List["Address"]] = relationship(#relationship()函数用于定义模型之间的关系，第一个参数为模型名，第二个参数为关系类型，back_populates参数用于指定反向引用的属性名。
        back_populates="user", cascade="all, delete-orphan"
    )
    def __repr__(self) -> str:#repr() 函数将对象转化为供解释器读取的形式。
        return f"User(id={self.id!r}, name={self.name!r}, fullname={self.fullname!r})"

class Address(Base):
    __tablename__ = "address"
    id: Mapped[int] = mapped_column(primary_key=True)
    email_address: Mapped[str] = mapped_column(String(50), nullable=False)
    user_id: Mapped[int] = mapped_column(ForeignKey("user_account.id"))
    user: Mapped["User"] = relationship(back_populates="addresses")
    def __repr__(self) -> str:
        return f"Address(id={self.id!r}, email_address={self.email_address!r})"
    
# Base.metadata.create_all(engine)
#插入操作
    """
    with Session(engine) as session: 创建了一个 SQLAlchemy 的 Session 对象，并将其赋值给 session。
    Session 对象实现了上下文管理协议，所以你可以在 with 语句的代码块中使用 session 来执行数据库操作，而不需要手动关闭 session,
    因为当离开 with 语句的代码块时,Session 对象的 __exit__ 方法会被自动调用，关闭数据库会话。
    总的来说，使用 with as 语句的目的是为了简化资源管理，确保资源在使用完毕后能被正确关闭和清理，避免资源泄露
    """
with Session(engine) as session:#Session()函数用于创建一个会话，会话用于执行SQL语句并管理事务，参数为数据库引擎。
    spongebob = User(
        name="spongebob", 
        fullname="Spongebob Squarepants",
        addresses=[
            Address(email_address="spongebob@sqlalchemy.org"),
        ]
    )
    sandy = User(
        name="sandy", 
        fullname="Sandy Cheeks",
        addresses=[
            Address(email_address="sandy@sqlalchmey.org"),
            Address(email_address="sandy@squirrelpower.org"),
        ],
    )
    partick = User(name="patrick", fullname="Patrick Star")
    #session.add_all([spongebob, sandy, partick])#add_all()函数用于将多个对象添加到会话中。
    #session.commit()#commit()函数用于提交事务，将会话中的所有操作作为一个事务提交到数据库中。

#简单查询操作
from sqlalchemy import select
with Session(engine) as session:
    stmt = select(User).where(User.name.in_(["spongebob", "sandy"]))#in_()函数用于判断某个字段的值是否在列表中。
    for user in session.scalars(stmt):#scalars()函数用于执行查询并返回结果集中。
        print(user)
#结果：
"""
User(id=1, name='spongebob', fullname='Spongebob Squarepants')
User(id=2, name='sandy', fullname='Sandy Cheeks')
"""
session = Session(engine)
#多表查询操作
stmt = (
    select(Address)
    .join(Address.user)
    .where(User.name == "sandy")
    .where(Address.email_address == "sandy_cheeks@sqlalchemy.org")
)
sandy_address = session.scalars(stmt).one()#one()函数用于返回结果集中的第一条记录。
#Address(id=2, email_address='sandy@sqlalchmey.org')

#更新操作
stmt = select(User).where(User.name == "patrick")
patrick = session.scalars(stmt).one()
#patrick=User(id=3, name='patrick', fullname='Patrick Star')
patrick.addresses.append(Address(email_address="patrickstar@sqlalchemy.org"))#添加新的邮箱地址
sandy_address.email_address = "sandy_cheeks@sqlalchemy.org"#修改邮箱地址
# session.commit()#提交事务

#删除操作
sandy = session.get(User,2)#get()函数用于根据主键获取记录。
sandy.addresses.remove(sandy_address)#删除一个邮箱地址
session.flush()#flush()在不提交事务的情况下，为更改的内容发出DELETE语句。
# 级联删除
session.delete(patrick)#删除用户,级联删除用户的邮箱地址
session.commit()#提交事务

2.使用orm的session

from sqlalchemy import create_engine

from sqlalchemy import text

from sqlalchemy.orm import Session

#与Mysql建立连接，用户名为remote，密码为123456，端口号为3306，数据库名称为vote  echo=True表示打印日志

engine = create_engine('mysql+pymysql://remote:qwer0487@192.168.198.186:3306/vote',echo=True)

session = Session(engine)

#使用text()方法执行原生sql语句

result = session.execute(text('select *from vote_announce'))

# print(result.all())

#使用映射关系获得数据from sqlalchemy import create_engine
from sqlalchemy import MetaData, ForeignKey
from sqlalchemy import Table, Column, Integer, String, ForeignKey
# #与Mysql建立连接，用户名为remote，密码为123456，端口号为3306，数据库名称为vote  echo=True表示打印日志
  engine = create_engine('mysql+pymysql://remote:qwer0487@192.168.198.186:3306/test',echo=True)
  

metadata_obj=MetaData()

user_table = Table(

		"user_account",#表名

		 metadata_obj,#assigns itself to a MetaData collection

		 #列对象

		Column("id", Integer, primary_key=True),

		Column("name", String(30)),

		Column("fullname", String(50)),

  

)

#简单的属性与方法

print(user_table.columns.id)#获取列对象

print(user_table.columns.keys())#获取所有列名

print(user_table.primary_key)#获取主键
#声明约束关系
adress_table = Table(

		 "address",#表名

		 metadata_obj,

		 Column("id", Integer, primary_key=True),

		 Column("user_id",ForeignKey("user_account.id"),nullable=False),#外键
		 Column("email_address", String(50), nullable=False),

)

#使用MetaData对象创建表

#Migration tools are usually appropriate,a schema management tool such as Alembic,最好使用迁移工具
metadata_obj.create_all(engine)

for row_dict in result.mappings():

    print(row_dict)

3.Table对象建立数据库元数据

from sqlalchemy import create_engine
from sqlalchemy import MetaData, ForeignKey,insert
from sqlalchemy import Table, Column, Integer, String, ForeignKey
# #与Mysql建立连接，用户名为remote，密码为123456，端口号为3306，数据库名称为vote  echo=True表示打印日志
engine = create_engine('mysql+pymysql://remote:qwer0487@192.168.198.186:3306/test',echo=True)
  

metadata_obj=MetaData()#创建一个MetaData对象

user_table = Table(
    "user_account",#表名
    metadata_obj,#assigns itself to a MetaData collection
    #列对象
    Column("id", Integer, primary_key=True),
    Column("name", String(30)),
    Column("fullname", String(50)),
)

#简单的属性与方法
print(user_table.columns.id)#获取列对象
# user_account.id
print(user_table.columns.keys())#获取所有列名
# ['id', 'name', 'fullname']
print(user_table.primary_key)#获取主键
# PrimaryKeyConstraint(Column('id', Integer(), table=<user_account>, primary_key=True, nullable=False))
#声明约束关系
adress_table = Table(
  "address",#表名
  metadata_obj,
  Column("id", Integer, primary_key=True),
  Column("user_id",ForeignKey("user_account.id"),nullable=False),#外键
  Column("email_address", String(50), nullable=False),
)
metadata_obj.create_all(engine)#发送数据库命令，创建表
# CREATE TABLE user_account (
#         id INTEGER NOT NULL AUTO_INCREMENT,
#         name VARCHAR(30),
#         fullname VARCHAR(50),
#         PRIMARY KEY (id)
# )


# 2024-01-04 14:20:10,877 INFO sqlalchemy.engine.Engine [no key 0.00059s] {}  
# 2024-01-04 14:20:10,889 INFO sqlalchemy.engine.Engine 
# CREATE TABLE address (
#         id INTEGER NOT NULL AUTO_INCREMENT,
#         user_id INTEGER NOT NULL,
#         email_address VARCHAR(50) NOT NULL,
#         PRIMARY KEY (id),
#         FOREIGN KEY(user_id) REFERENCES user_account (id)
# )


# 2024-01-04 14:20:10,890 INFO sqlalchemy.engine.Engine [no key 0.00092s] {}  
# 2024-01-04 14:20:10,906 INFO sqlalchemy.engine.Engine COMMIT
# 1.建立数据库引擎，使用引擎自带的connect
```python
from sqlalchemy import create_engine

from sqlalchemy import text

#与Mysql建立连接，用户名为remote，密码为123456，端口号为3306，数据库名称为vote  echo=True表示打印日志

engine = create_engine('mysql+pymysql://remote:qwer0487@192.168.198.186:3306/vote',echo=True)

conn = engine.connect()

#使用text()方法执行原生sql语句

result = conn.execute(text('select *from vote_announce'))

# print(result.all())

#使用映射关系获得数据

for row_dict in result.mappings():

    print(row_dict)

4.sqlalchemy建立数据库引擎

from sqlalchemy import create_engine

from sqlalchemy import text

#与Mysql建立连接，用户名为remote，密码为123456，端口号为3306，数据库名称为vote  echo=True表示打印日志

engine = create_engine('mysql+pymysql://remote:qwer0487@192.168.198.186:3306/vote',echo=True)

conn = engine.connect()

#使用text()方法执行原生sql语句

result = conn.execute(text('select *from vote_announce'))

# print(result.all())

#使用映射关系获得数据

for row_dict in result.mappings():

    print(row_dict)

5.Alembic数据库迁移的步骤

基础步骤
  • 安装 Alembic：
pip install alembic
  • 创建数据库模型： 在你的 Python 项目中创建一个文件，用于定义数据库模型。这通常是使用 SQLAlchemy 来定义的。
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'

    id = Column(Integer, primary_key=True)
    username = Column(String, nullable=False)
    email = Column(String, nullable=False)
  • 配置 Alembic： 使用 Alembic 初始化配置。
alembic init alembic

这将在项目根目录下创建一个名为 alembic 的目录，包含配置文件和其他必要的文件。

  • 编辑 Alembic 配置文件：

编辑 alembic.ini 文件，配置数据库连接字符串。

sqlalchemy.url = driver://user:password@localhost/dbname
  • 编辑 alembic/env.py 文件，将数据库模型导入并与 Base 绑定。
from your_project.models import Base
target_metadata = Base.metadata
  • 创建迁移脚本： 这将自动生成一个迁移脚本，描述了数据库模型的创建过程。
alembic revision --autogenerate -m "create tables"
  • 升级数据库：
alembic upgrade head

注意

易错点注释
文章作者: Hkini
文章链接: https://www.hkini.top/posts/4270.html
版权声明: 本博客所有文章除特別声明外，均采用 CC BY 4.0 许可协议。转载请注明来源 Hkini !
Python
赏
评论
  目录