我将 sqlalchemy 与 sqlite 和 aiosqlite 驱动程序一起使用。
我创建了两个表,如下所示:
class AccountsModel(Base):
__tablename__ = 'accounts'
id: Mapped[intpk]
email_id: Mapped[int] = mapped_column(
ForeignKey('emails.id', ondelete='CASCADE')
)
new_email_id: Mapped[int | None] = mapped_column(
ForeignKey('new_emails.id', ondelete='CASCADE'),
nullable=True
)
email: Mapped['EmailsModel'] = relationship(
back_populates='account',
single_parent=True,
cascade='all, delete-orphan',
lazy='joined'
)
new_email: Mapped['NewEmailsModel'] = relationship(
back_populates='account',
single_parent=True,
cascade='all, delete-orphan',
lazy='joined'
)
class EmailsModel(Base):
__tablename__ = 'emails'
id: Mapped[intpk]
email: Mapped[str_255] = mapped_column(unique=True)
account: Mapped['AccountsModel'] = relationship(
back_populates='email',
cascade='all, delete-orphan'
)
发动机:
class SessionMaker:
def __init__(self, dsn: str):
self._engine = create_async_engine(dsn)
@staticmethod
@event.listens_for(Engine, 'connect')
def _set_sqlite_pragma(connection: Any, record: '_ConnectionRecord'):
cursor = connection.cursor()
cursor.execute('PRAGMA foreign_keys=ON')
cursor.close()
async def create_tables(self, drop=False) -> None:
async with self._engine.begin() as conn:
if drop:
await conn.run_sync(Base.metadata.drop_all)
await conn.run_sync(Base.metadata.create_all)
@property
def session(self) -> async_sessionmaker[AsyncSession]:
if not hasattr(self, '_session'):
self._session = async_sessionmaker(self._engine)
return self._session
查询:
async def delete(
self,
*,
filter_by: Dict[str, Any]
) -> None:
async with self._session() as session:
stmt = (
delete(self._model)
.filter_by(**filter_by)
)
await session.execute(stmt)
await session.commit()
async def delete_by_id(
self,
*,
id: int,
) -> None:
await self.delete(filter_by=dict(id=id))
await MY_MODEL.delete_by_id(MY_ID)
当我执行上面的查询时,我看到以下输出:
BEGIN (implicit)
DELETE FROM accounts WHERE accounts.id = ?
[cached since 20.71s ago] (3,)
COMMIT
创建语句:
CREATE TABLE accounts (
id INTEGER NOT NULL,
email_id INTEGER NOT NULL,
new_email_id INTEGER,
PRIMARY KEY (id),
FOREIGN KEY(email_id) REFERENCES emails (id) ON DELETE CASCADE,
FOREIGN KEY(new_email_id) REFERENCES new_emails (id) ON DELETE CASCADE
)
CREATE TABLE emails (
id INTEGER NOT NULL,
email VARCHAR(255) NOT NULL,
PRIMARY KEY (id),
UNIQUE (email)
)
我在 stackoverflow 上发现了类似的问题,但仍然不明白我的代码出了什么问题。我很高兴获得任何帮助,谢谢。
DELETE FROM accounts WHERE accounts.id = ?
瀑布般的瀑布,向下流淌。
当您编写外键时,它会在 LOWER 表(子表)上进行编码,并指定对 HIGHER 表(父表)的引用。因此
ON DELETE CASCADE
,如果删除引用的父电子邮件/new_emails,则在帐户上编码将导致删除帐户行。
通常,您会期望电子邮件由某人/某物拥有(发送/接收),因此电子邮件/new_email 是帐户的子帐户。因此,也许您应该在 emails/new_emails 表上编码外键。然后删除帐户将会级联删除到 emails/new_emails。