博客 / 詳情

返回

對於 SQLite 使用 SQLAlchemy WriteOnlyMapped 時，“刪除”級聯行為失效的解決方案

背景

WriteOnlyMapped 使用 lazy="write_only" 加載策略，該策略不在內存中加載集合的內容，只在寫入時進行操作。這對於包含大量數據的集合特別有用，因為不會因加載所有數據而導致內存佔用過高【1】。

問題描述

from sqlalchemy import create_engine, ForeignKey
from sqlalchemy.orm import (
    DeclarativeBase,
    WriteOnlyMapped,
    Mapped,
    sessionmaker,
    relationship,
    mapped_column
)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = 'user'
    id: Mapped[int] = mapped_column(primary_key=True)
    addresses: WriteOnlyMapped['Address'] = relationship(
        cascade="all, delete-orphan", passive_deletes=True)

    def __repr__(self):
        return f'<User {self.id}>'


class Address(Base):
    __tablename__ = 'address'
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(
        ForeignKey('user.id', ondelete='cascade'))

    def __repr__(self):
        return f'<Address {self.id}>'


engine = create_engine('sqlite:///:memory:')
Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

user = User()
session.add(user)
user = session.get(User, 1)
address1 = Address(user_id=user.id)
address2 = Address(user_id=user.id)
session.add(address1)
session.add(address2)
session.commit()

print(session.query(User).all())
print(session.query(Address).all())

session.delete(user)
session.commit()

print('\ndelete user')
print(session.query(User).all())
print(session.query(Address).all())

上述代碼時官方文檔示例【1】的簡化，期望的效果是刪除一個 user 後，該 user 的 address 會被自動刪除：

[<User 1>]
[<Address 1>, <Address 2>]

delete user
[]
[]

但是由於使用了 SQLite，實際是該 user 的 address 並沒有被自動刪除：

[<User 1>]
[<Address 1>, <Address 2>]

delete user
[]
[<Address 1>, <Address 2>]

解決方案

使用 SQLite 時，必須顯式啓用外鍵支持【2】【3】:

import re
import sqlite3
from sqlalchemy import Engine, event

def is_sqlite(db_uri):
    """檢查數據庫 URI 是否為 SQLite"""
    return bool(re.match(r'sqlite:///', db_uri))

# 每次建立新連接時執行 set_sqlite_pragma（僅在使用 SQLite 時）
if is_sqlite(db_uri):
    @event.listens_for(Engine, 'connect')
    def set_sqlite_pragma(dbapi_connection: sqlite3.Connection, _):
        """啓用 SQLite 的外鍵約束"""
        cursor = dbapi_connection.cursor()
        cursor.execute('PRAGMA foreign_keys=ON')
        cursor.close()

完整示例代碼

import re
import sqlite3
from sqlalchemy import (
    ForeignKey,
    Engine,
    event,
    create_engine
)
from sqlalchemy.orm import (
    DeclarativeBase,
    WriteOnlyMapped,
    Mapped,
    sessionmaker,
    relationship,
    mapped_column
)


class Base(DeclarativeBase):
    pass


class User(Base):
    __tablename__ = 'user'
    id: Mapped[int] = mapped_column(primary_key=True)
    addresses: WriteOnlyMapped['Address'] = relationship(
        cascade="all, delete-orphan", passive_deletes=True)

    def __repr__(self):
        return f'<User {self.id}>'


class Address(Base):
    __tablename__ = 'address'
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(
        ForeignKey('user.id', ondelete='cascade'))

    def __repr__(self):
        return f'<Address {self.id}>'


def is_sqlite(db_uri):
    """檢查數據庫 URI 是否為 SQLite"""
    return bool(re.match(r'sqlite:///', db_uri))

db_uri = 'sqlite:///:memory:'
engine = create_engine(db_uri)

# 每次建立新連接時執行 set_sqlite_pragma（僅在使用 SQLite 時）
if is_sqlite(db_uri):
    @event.listens_for(Engine, 'connect')
    def set_sqlite_pragma(dbapi_connection: sqlite3.Connection, _):
        """啓用 SQLite 的外鍵約束"""
        cursor = dbapi_connection.cursor()
        cursor.execute('PRAGMA foreign_keys=ON')
        cursor.close()

Base.metadata.create_all(engine)

Session = sessionmaker(bind=engine)
session = Session()

user = User()
session.add(user)
user = session.get(User, 1)
address1 = Address(user_id=user.id)
address2 = Address(user_id=user.id)
session.add(address1)
session.add(address2)
session.commit()

print(session.query(User).all())
print(session.query(Address).all())

session.delete(user)
session.commit()

print('\ndelete user')
print(session.query(User).all())
print(session.query(Address).all())

完整示例代碼（Flask-SQLAlchemy）

import re
import sqlite3
from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from sqlalchemy import (
    ForeignKey,
    Engine,
    event
)
from sqlalchemy.orm import (
    DeclarativeBase,
    Mapped,
    WriteOnlyMapped,
    relationship,
    mapped_column
)


class Base(DeclarativeBase):
    pass


app = Flask(__name__)
app.config['SQLALCHEMY_DATABASE_URI'] = 'sqlite:///:memory:'
db = SQLAlchemy(model_class=Base)
db.init_app(app)


class User(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    addresses: WriteOnlyMapped['Address'] = relationship(
        cascade="all, delete-orphan", passive_deletes=True)

    def __repr__(self):
        return f'<User {self.id}>'


class Address(db.Model):
    id: Mapped[int] = mapped_column(primary_key=True)
    user_id: Mapped[int] = mapped_column(
        ForeignKey('user.id', ondelete='cascade'))

    def __repr__(self):
        return f'<Address {self.id}>'


def is_sqlite(db_uri):
    """檢查數據庫 URI 是否為 SQLite"""
    return bool(re.match(r'sqlite:///', db_uri))


# 每次建立新連接時執行 set_sqlite_pragma（僅在使用 SQLite 時）
if is_sqlite(app.config['SQLALCHEMY_DATABASE_URI']):
    @event.listens_for(Engine, 'connect')
    def set_sqlite_pragma(dbapi_connection: sqlite3.Connection, _):
        """啓用 SQLite 的外鍵約束"""
        cursor = dbapi_connection.cursor()
        cursor.execute('PRAGMA foreign_keys=ON')
        cursor.close()


with app.app_context():
    db.create_all()

    user = User()
    db.session.add(user)
    user = db.session.get(User, 1)
    address1 = Address(user_id=user.id)
    address2 = Address(user_id=user.id)
    db.session.add(address1)
    db.session.add(address2)
    db.session.commit()

    print(db.session.query(User).all())
    print(db.session.query(Address).all())

    db.session.delete(user)
    db.session.commit()

    print('\ndelete user')
    print(db.session.query(User).all())
    print(db.session.query(Address).all())
flask-sqlalchemy , sqlalchemy , sqlite , Python
user avatar
0 位用戶收藏了這個故事！

發佈 評論

Some HTML is okay.