Атрибут identity_map класса AsyncSession

Атрибут identity_map класса AsyncSession возвращает объект карты идентичности (identity map). Карта идентичности представляет собой словареподобную структуру, в которой хранятся все объекты ORM, загруженные в текущей сессии, с ключом в виде кортежа из класса и первичного ключа. Этот атрибут позволяет проверить, загружен ли уже определенный объект в сессию, и получить к нему доступ напрямую без выполнения дополнительных SQL-запросов.

Карта идентичности обеспечивает гарантию того, что в рамках одной сессии для одной строки базы данных существует ровно один Python-объект. При повторном запросе той же записи SQLAlchemy вернет уже существующий объект из карты, а не создаст новый.

Синтаксис

async_session.identity_map

Пример

Давайте создадим асинхронную сессию, добавим объект и проверим содержимое карты идентичности:

import asyncio
from sqlalchemy import Integer, String
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'
	
	id: Mapped[int] = mapped_column(Integer, primary_key=True)
	title: Mapped[str] = mapped_column(String)

async def main():
	engine = create_async_engine('sqlite+aiosqlite:///:memory:')
	
	async with engine.begin() as conn:
		await conn.run_sync(Base.metadata.create_all)
	
	async with AsyncSession(engine) as session:
		article = Article(title='article')
		session.add(article)
		await session.flush()
		
		print(session.identity_map)
		print(len(session.identity_map))

asyncio.run(main())

Результат выполнения кода:

"<IdentitySet at 0x...; 1 objects>"
"1"

Пример

Давайте проверим, содержится ли объект в карте идентичности по его ключу:

<+python+> import asyncio from sqlalchemy import Integer, String from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession class Base(DeclarativeBase): pass class Article(Base): __tablename__ = 'articles' id: Mapped[int] = mapped_column(Integer, primary_key=True) title: Mapped[str] = mapped_column(String) async def main(): engine = create_async_engine('sqlite+aiosqlite:///:memory:') async with engine.begin() as conn: await conn.run_sync(Base.metadata.create_all) async with AsyncSession(engine) as session: article = Article(title='article') session.add(article) await session.flush() key = (Article, article.id) print(key in session.identity_map) print(session.identity_map[key] is article) asyncio.run(main()) <-python+>

Результат выполнения кода:

"True"
"True"

Пример

Давайте убедимся, что при повторной загрузке той же записи SQLAlchemy возвращает тот же объект из карты идентичности:

import asyncio
from sqlalchemy import Integer, String, select
from sqlalchemy.orm import DeclarativeBase, Mapped, mapped_column
from sqlalchemy.ext.asyncio import create_async_engine, AsyncSession

class Base(DeclarativeBase):
	pass

class Article(Base):
	__tablename__ = 'articles'
	
	id: Mapped[int] = mapped_column(Integer, primary_key=True)
	title: Mapped[str] = mapped_column(String)

async def main():
	engine = create_async_engine('sqlite+aiosqlite:///:memory:')
	
	async with engine.begin() as conn:
		await conn.run_sync(Base.metadata.create_all)
	
	async with AsyncSession(engine) as session:
		article = Article(title='article')
		session.add(article)
		await session.commit()
		
		res1 = await session.get(Article, article.id)
		res2 = await session.get(Article, article.id)
		
		print(res1 is res2)
		print(len(session.identity_map))

asyncio.run(main())

Результат выполнения кода:

"True"
"1"

Смотрите также

класс AsyncSession,
который представляет асинхронную сессию
метод get,
который загружает объект по первичному ключу
метод expunge,
который удаляет объект из сессии и карты идентичности
метод expunge_all,
который очищает сессию и карту идентичности полностью