Method execute

ormar/databases/query_executor.py:59–87

Execute a query (INSERT, UPDATE, DELETE). :param query: SQLAlchemy query expression :return: For INSERT, the inserted primary key or ``None`` if the backend cannot return one (e.g. Oracle MySQL inserting into a non-AUTO_INCREMENT pk with a server def

(self, query: Executable)

        return result.scalar()

    async def execute(self, query: Executable) -> Any:
        """
        Execute a query (INSERT, UPDATE, DELETE).

        :param query: SQLAlchemy query expression
        :return: For INSERT, the inserted primary key or ``None`` if the backend
            cannot return one (e.g. Oracle MySQL inserting into a
            non-AUTO_INCREMENT pk with a server default — no RETURNING support).
            For UPDATE/DELETE, the row count.
        """
        result: CursorResult[Any] = await self._connection.execute(query)

        # For INSERT queries, try to get the inserted primary key via the
        # dialect's best-available mechanism (RETURNING on PostgreSQL / SQLite
        # 3.35+ / MariaDB 10.5+, LAST_INSERT_ID() on MySQL AUTO_INCREMENT).
        # Do NOT fall back to rowcount here: rowcount is not a pk, and
        # returning it would silently corrupt `Model.pk` in `save()`.
        if result.context and result.context.isinsert:
            if result.inserted_primary_key:
                pk_value = result.inserted_primary_key[0]
                if pk_value is not None:
                    return pk_value

            if hasattr(result, "lastrowid") and result.lastrowid:  # pragma: no cover
                return result.lastrowid

            return None  # pragma: no cover

        return result.rowcount if result.rowcount is not None else 0

    async def execute_many(
        self, query: Union[Executable, str], values: Sequence[Mapping[str, Any]]