prefer-bulk-upsert · Python

Why

Serial singleton calls and eager coroutine fan-out can repeat database or remote work. A source-declared compatible bulk API identifies an opportunity to review batching, including bulk implementations that repeat backend singleton calls.

Fix

Collect inputs and use the declared bulk API in bounded batches after reviewing missing rows, cardinality, duplicate keys, ordering, failure atomicity, concurrency, cancellation, backpressure, tenant scope, and transaction or snapshot behavior. Preserve deliberately per-row transactions or effects with an exact suppression.

Examples

Before — flagged Review repeated reads against an available bulk API

class UserStore:
    async def get(self, key: str) -> User | None: ...
    async def get_many(self, keys: list[str]) -> list[User]: ...


async def load(store: UserStore, keys: list[str]):
    return [await store.get(key) for key in keys]

After — preferred Read the batch after reviewing missing rows and ordering

class UserStore:
    async def get(self, key: str) -> User | None: ...
    async def get_many(self, keys: list[str]) -> list[User]: ...


async def load(store: UserStore, keys: list[str]):
    return await store.get_many(keys)

Before — flagged Concurrency still schedules one operation per input

import asyncio


class UserStore:
    async def get(self, key: str) -> User | None: ...
    async def get_many(self, keys: list[str]) -> list[User]: ...


async def load(store: UserStore, keys: list[str]):
    return await asyncio.gather(*(store.get(key) for key in keys))

After — preferred Use bounded bulk reads after reviewing concurrency and failures

class UserStore:
    async def get(self, key: str) -> User | None: ...
    async def get_many(self, keys: list[str]) -> list[User]: ...


async def load(store: UserStore, keys: list[str]):
    return await store.get_many(keys)

Before — flagged Collect writes for an available bulk command

class UserStore:
    async def upsert(self, user: UserInput) -> User: ...
    async def upsert_many(self, users: list[UserInput]) -> list[User]: ...


async def save(store: UserStore, users: list[UserInput]):
    for user in users:
        await store.upsert(user)

After — preferred Use the existing bulk command

class UserStore:
    async def upsert(self, user: UserInput) -> User: ...
    async def upsert_many(self, users: list[UserInput]) -> list[User]: ...


async def save(store: UserStore, users: list[UserInput]):
    await store.upsert_many(users)

Before — flagged Review a comprehension of singleton writes

class UserStore:
    async def upsert(self, user: UserInput) -> User: ...
    async def upsert_many(self, users: list[UserInput]) -> list[User]: ...


async def save(store: UserStore, users: list[UserInput]):
    return [await store.upsert(user) for user in users]

After — preferred Return bulk results after reviewing cardinality and failures

class UserStore:
    async def upsert(self, user: UserInput) -> User: ...
    async def upsert_many(self, users: list[UserInput]) -> list[User]: ...


async def save(store: UserStore, users: list[UserInput]):
    return await store.upsert_many(users)