How can i delete records from ManyToMany by several ids like [1,2,3..]

[shurshilov]

How can i delete records from ManyToMany by several ids like [1,2,3..]
because delete onebyone very slowly

[vlakius]

You could execute the delete concurrently using asyncio.gather:

ids = [1,2,3,4]
delete_tasks = [YourModel.get(id=id).delete() for id in ids]
await asyncio.gather(*delete_tasks)

[alexanderankin]

You can do something like:

e3 = await ExampleModel.create(name="e3")
await e3.tags.add(t1, t2, t3)
await e3.tags.remove(t2, t3)

where tags is a ManyToManyRelation (from tortoise.fields)

here is a self contained code sample, which requires (pip install testcontainers tortoise-orm pytest pytest-asyncio asyncpg)

# pip install testcontainers tortoise-orm pytest pytest-asyncio asyncpg
from dataclasses import dataclass
from re import sub
from uuid import UUID

import pytest
import pytest_asyncio
from testcontainers.postgres import PostgresContainer
from tortoise import Model, Tortoise
from tortoise.fields import UUIDField, CharField, TextField, ManyToManyField, ManyToManyRelation


class ExampleModel(Model):
    id: UUID = UUIDField(pk=True, null=False)
    name: str = CharField(max_length=255, unique=True, null=False)
    comment: str = TextField(null=True)

    tags: ManyToManyRelation["TagModel"] = ManyToManyField(
        model_name="app.TagModel",
        related_name="examples",
        through="example_tag",
        forward_key="tag_id",
        backward_key="example_id",
        default=(((),)[0]),  # empty tuple
    )

    class Meta:
        table = "example"


class TagModel(Model):
    id: UUID = UUIDField(pk=True, null=False)
    name: str = CharField(max_length=255, unique=True, null=False)
    comment: str = TextField(null=True)

    examples: ManyToManyRelation[ExampleModel]

    class Meta:
        table = "tag"


@dataclass
class TextCtx:
    pg: PostgresContainer
    url: str


@pytest_asyncio.fixture(scope="session")
async def test_context():
    pg = PostgresContainer("postgres:16-alpine", driver=None)
    pg.start()
    url = pg.get_connection_url()
    asyncpg_url = sub(r"^(postgres.{0,20}?)(?=://)", "asyncpg", url)

    import asyncpg
    conn = await asyncpg.connect(url)
    await conn.execute("""
        CREATE TABLE IF NOT EXISTS "example"
        (
            "id"      UUID         NOT NULL PRIMARY KEY,
            "name"    VARCHAR(255) NOT NULL UNIQUE,
            "comment" TEXT         NULL
        );
        CREATE TABLE IF NOT EXISTS "tag"
        (
            "id"      UUID         NOT NULL PRIMARY KEY,
            "name"    VARCHAR(255) NOT NULL UNIQUE,
            "comment" TEXT         NULL
        );
        CREATE TABLE "example_tag"
        (
            "example_id" UUID NOT NULL REFERENCES "example" ("id") ON DELETE CASCADE,
            "tag_id"     UUID NOT NULL REFERENCES "tag" ("id") ON DELETE CASCADE
        );
    """)
    await conn.close()

    await Tortoise.init({
        "connections": {"default": asyncpg_url},
        "apps": {
            "app": {
                "models": ["example_delete_many_to_many_tortoise_orm"],
                "default_connection": "default"
            }
        }
    })

    yield TextCtx(pg=pg, url=url)


@pytest.mark.asyncio
async def test_example(test_context):
    example = await ExampleModel.create(name="example")
    assert example.id is not None

    e1 = await ExampleModel.create(name="e1")
    e2 = await ExampleModel.create(name="e2")
    e3 = await ExampleModel.create(name="e3")

    t1 = await TagModel.create(name="t1")
    t2 = await TagModel.create(name="t2")
    t3 = await TagModel.create(name="t3")

    await e1.tags.add(t1)
    await e2.tags.add(t1, t2)
    await e3.tags.add(t1, t2, t3)

    e2_new = await ExampleModel.get(id=e2.id).prefetch_related("tags")
    tags = e2_new.tags
    print(tags.related_objects)
    assert len(tags.related_objects) == 2
    assert all(isinstance(t, TagModel) for t in tags.related_objects)

    await e3.tags.remove(t2, t3)

    e3_new = await ExampleModel.get(id=e3.id).prefetch_related("tags")
    tags = e3_new.tags
    assert len(tags.related_objects) == 1
    assert tags.related_objects[0].id == t1.id