SQLAlchemy + asyncpg doesn't catch deferred exception

Question

I am now using SQLAlchemy 1.4, asyncpg and FastAPI, and I wrote the code like this:

    try:
        cr: sa.engine.CursorResult = await conn.execute(stmt)
        return schemas.UserGroup(
            **user_group_dict,
            id=cr.inserted_primary_key[0],
        )
    except sa.exc.IntegrityError:
        raise exceptions.conflict_exception()

UserGroup table references User and Group tables and has a unique key constraint (user, group)

user_groups = sa.Table(
    "auth_user_groups",
    metadata,
    sa.Column(
        "id",
        sa.BigInteger,
        primary_key=True,
        index=True,
    ),
    sa.Column(
        "user_id",
        sa.BigInteger,
        sa.ForeignKey("auth_user.id"),
    ),
    sa.Column(
        "group_id",
        sa.BigInteger,
        sa.ForeignKey("auth_group.id"),
    ),
    sa.UniqueConstraint("user_id", "group_id"),
)

CREATE TABLE IF NOT EXISTS public.auth_user_groups
(
    id bigint NOT NULL DEFAULT nextval('auth_user_groups_id_seq'::regclass),
    user_id integer NOT NULL,
    group_id integer NOT NULL,
    CONSTRAINT auth_user_groups_pkey PRIMARY KEY (id),
    CONSTRAINT auth_user_groups_user_id_group_id_94350c0c_uniq UNIQUE (user_id, group_id),
    CONSTRAINT auth_user_groups_group_id_97559544_fk_auth_group_id FOREIGN KEY (group_id)
        REFERENCES public.auth_group (id) MATCH SIMPLE
        ON UPDATE NO ACTION
        ON DELETE NO ACTION
        DEFERRABLE INITIALLY DEFERRED,
    CONSTRAINT auth_user_groups_user_id_6a12ed8b_fk_auth_user_id FOREIGN KEY (user_id)
        REFERENCES public.auth_user (id) MATCH SIMPLE
        ON UPDATE NO ACTION
        ON DELETE NO ACTION
        DEFERRABLE INITIALLY DEFERRED
)

When I try to insert a duplicated record, it works fine.

However, when I try to insert a record with fk which does not exist in User and Group, I cannot catch the exception.

2022-06-14 01:02:50,978 INFO sqlalchemy.engine.Engine COMMIT
ERROR:    Exception in ASGI application
Traceback (most recent call last):
  File "/home/test/projects/fastapi-from-dj/venv/lib/python3.10/site-packages/uvicorn/protocols/http/h11_impl.py", line 366, in run_asgi
    result = await app(self.scope, self.receive, self.send)
  File "/home/test/projects/fastapi-from-dj/venv/lib/python3.10/site-packages/uvicorn/middleware/proxy_headers.py", line 75, in __call__
    return await self.app(scope, receive, send)
...
  File "/home/test/projects/fastapi-from-dj/venv/lib/python3.10/site-packages/sqlalchemy/engine/default.py", line 685, in do_commit
    dbapi_connection.commit()
  File "/home/test/projects/fastapi-from-dj/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 741, in commit
    self._handle_exception(error)
  File "/home/test/projects/fastapi-from-dj/venv/lib/python3.10/site-packages/sqlalchemy/dialects/postgresql/asyncpg.py", line 682, in _handle_exception
    raise translated_error from error
sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) <class 'asyncpg.exceptions.ForeignKeyViolationError'>: insert or update on table "auth_user_groups" violates foreign key constraint "auth_user_groups_user_id_6a12ed8b_fk_auth_user_id"
DETAIL:  Key (user_id)=(15) is not present in table "auth_user".
(Background on this error at: https://sqlalche.me/e/14/gkpj)

Of course, I tried to catch exception with asyncpg.exceptions.ForeignKeyViolationError and Exception as e, but I failed.

Thank you.

Answer 1

Thanks to @Gord Thompson's comment

I was able to solve my problem by altering the table FK constraints:

ALTER TABLE IF EXISTS public.auth_user_groups
    DROP CONSTRAINT auth_user_groups_user_id_6a12ed8b_fk_auth_user_id;

ALTER TABLE IF EXISTS public.auth_user_groups
    ADD CONSTRAINT auth_user_groups_user_id_6a12ed8b_fk_auth_user_id FOREIGN KEY (user_id)
    REFERENCES public.auth_user (id) MATCH SIMPLE
    ON UPDATE NO ACTION
    ON DELETE NO ACTION
    NOT DEFERRABLE;

As a result, now I have a table like this:

CREATE TABLE IF NOT EXISTS public.auth_user_groups
(
    id bigint NOT NULL DEFAULT nextval('auth_user_groups_id_seq'::regclass),
    user_id integer NOT NULL,
    group_id integer NOT NULL,
    CONSTRAINT auth_user_groups_pkey PRIMARY KEY (id),
    CONSTRAINT auth_user_groups_user_id_group_id_94350c0c_uniq UNIQUE (user_id, group_id),
    CONSTRAINT auth_user_groups_group_id_97559544_fk_auth_group_id FOREIGN KEY (group_id)
        REFERENCES public.auth_group (id) MATCH SIMPLE
        ON UPDATE NO ACTION
        ON DELETE NO ACTION,
    CONSTRAINT auth_user_groups_user_id_6a12ed8b_fk_auth_user_id FOREIGN KEY (user_id)
        REFERENCES public.auth_user (id) MATCH SIMPLE
        ON UPDATE NO ACTION
        ON DELETE NO ACTION
)

It does not have the line: DEFERRABLE INITIALLY DEFERRED

I am now using the tables which were migrated by Django so that FastAPI uses them as a REST API.

I believe that the following answer link will be useful for Django users:

How can I set a table constraint "deferrable initially deferred" in django model?

Thank you.