<h3 dir="auto">Describe the bug</h3>
<p dir="auto">Hi,<br>
I'm using imperative mapping to specify a custom mapping of a DB table to my domain model. In detail, I want that on the DB the primary key is an integer ID field, while on the codebase the entity model maps its ID to the external_id (UUID) field of the table.<br>
I want that from a codebase point of view the UUID is the primary key, not the integer field.</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content="@dataclass(frozen=False)
class LicenseSetting:
    account_id: UUID
    reassign_licenses_on_license_pool_expiration: bool
    allowed_reassigned_percentage: int
    id: int = field(default_factory=uuid.uuid4) # I want to map it to the database external_id field

metadata = get_db_metadata()
mapper_registry = get_mapper_registry()

license_setting_table = Table(
    'license_setting',
    metadata,
    Column('id', Integer(), primary_key=True),
    Column('external_id', UUID(as_uuid=True), unique=True, nullable=False, default=uuid.uuid4),
    Column('reassign_licenses_on_license_pool_expiration', Boolean(), default=False, nullable=False),
    Column('allowed_reassigned_percentage', Integer(), default=0, nullable=False),
    Column('account_id', UUID(as_uuid=True), unique=True, nullable=False)
)

mapper_registry.map_imperatively(
    LicenseSetting,
    license_setting_table,
    primary_key=[license_setting_table.c[&quot;external_id&quot;]], # here I specify I want use the field external_id as primary key
    exclude_properties=[&quot;id&quot;], # real 'id' of the table is not mapped to my domain entity
    properties={
        'id': license_setting_table.c[&quot;external_id&quot;]  # here I specify I want to map db 'external_id' to entity 'id'
    })"><pre class="notranslate"><span class="pl-en">@<span class="pl-en">dataclass</span>(<span class="pl-s1">frozen</span><span class="pl-c1">=</span><span class="pl-c1">False</span>)</span>
<span class="pl-k">class</span> <span class="pl-v">LicenseSetting</span>:
    <span class="pl-s1">account_id</span>: <span class="pl-v">UUID</span>
    <span class="pl-s1">reassign_licenses_on_license_pool_expiration</span>: <span class="pl-s1">bool</span>
    <span class="pl-s1">allowed_reassigned_percentage</span>: <span class="pl-s1">int</span>
    <span class="pl-s1">id</span>: <span class="pl-s1">int</span> <span class="pl-c1">=</span> <span class="pl-en">field</span>(<span class="pl-s1">default_factory</span><span class="pl-c1">=</span><span class="pl-s1">uuid</span>.<span class="pl-s1">uuid4</span>) <span class="pl-c"># I want to map it to the database external_id field</span>

<span class="pl-s1">metadata</span> <span class="pl-c1">=</span> <span class="pl-en">get_db_metadata</span>()
<span class="pl-s1">mapper_registry</span> <span class="pl-c1">=</span> <span class="pl-en">get_mapper_registry</span>()

<span class="pl-s1">license_setting_table</span> <span class="pl-c1">=</span> <span class="pl-v">Table</span>(
    <span class="pl-s">'license_setting'</span>,
    <span class="pl-s1">metadata</span>,
    <span class="pl-v">Column</span>(<span class="pl-s">'id'</span>, <span class="pl-v">Integer</span>(), <span class="pl-s1">primary_key</span><span class="pl-c1">=</span><span class="pl-c1">True</span>),
    <span class="pl-v">Column</span>(<span class="pl-s">'external_id'</span>, <span class="pl-v">UUID</span>(<span class="pl-s1">as_uuid</span><span class="pl-c1">=</span><span class="pl-c1">True</span>), <span class="pl-s1">unique</span><span class="pl-c1">=</span><span class="pl-c1">True</span>, <span class="pl-s1">nullable</span><span class="pl-c1">=</span><span class="pl-c1">False</span>, <span class="pl-s1">default</span><span class="pl-c1">=</span><span class="pl-s1">uuid</span>.<span class="pl-s1">uuid4</span>),
    <span class="pl-v">Column</span>(<span class="pl-s">'reassign_licenses_on_license_pool_expiration'</span>, <span class="pl-v">Boolean</span>(), <span class="pl-s1">default</span><span class="pl-c1">=</span><span class="pl-c1">False</span>, <span class="pl-s1">nullable</span><span class="pl-c1">=</span><span class="pl-c1">False</span>),
    <span class="pl-v">Column</span>(<span class="pl-s">'allowed_reassigned_percentage'</span>, <span class="pl-v">Integer</span>(), <span class="pl-s1">default</span><span class="pl-c1">=</span><span class="pl-c1">0</span>, <span class="pl-s1">nullable</span><span class="pl-c1">=</span><span class="pl-c1">False</span>),
    <span class="pl-v">Column</span>(<span class="pl-s">'account_id'</span>, <span class="pl-v">UUID</span>(<span class="pl-s1">as_uuid</span><span class="pl-c1">=</span><span class="pl-c1">True</span>), <span class="pl-s1">unique</span><span class="pl-c1">=</span><span class="pl-c1">True</span>, <span class="pl-s1">nullable</span><span class="pl-c1">=</span><span class="pl-c1">False</span>)
)

<span class="pl-s1">mapper_registry</span>.<span class="pl-en">map_imperatively</span>(
    <span class="pl-v">LicenseSetting</span>,
    <span class="pl-s1">license_setting_table</span>,
    <span class="pl-s1">primary_key</span><span class="pl-c1">=</span>[<span class="pl-s1">license_setting_table</span>.<span class="pl-s1">c</span>[<span class="pl-s">"external_id"</span>]], <span class="pl-c"># here I specify I want use the field external_id as primary key</span>
    <span class="pl-s1">exclude_properties</span><span class="pl-c1">=</span>[<span class="pl-s">"id"</span>], <span class="pl-c"># real 'id' of the table is not mapped to my domain entity</span>
    <span class="pl-s1">properties</span><span class="pl-c1">=</span>{
        <span class="pl-s">'id'</span>: <span class="pl-s1">license_setting_table</span>.<span class="pl-s1">c</span>[<span class="pl-s">"external_id"</span>]  <span class="pl-c"># here I specify I want to map db 'external_id' to entity 'id'</span>
    })</pre></div>
<p dir="auto">The issue is that when I try to add two domain instances with the same identifier I receive an integrity error. It seems that the ORM is not able to detect that I'm using the external_id as the primary key generating a single insert statement.<br>
Instead, it tries two add two rows with the same external_id and two different ids. Since on the external_id there is a unique constraint, I receive an integrity error.</p>
<p dir="auto">The error I receive:</p>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="self = &lt;sqlalchemy.dialects.postgresql.asyncpg.AsyncAdapt_asyncpg_connection object at 0x10a30be40&gt;
error = UniqueViolationError('duplicate key value violates unique constraint &quot;license_setting_external_id_key&quot;')

    def _handle_exception(self, error):
        if self._connection.is_closed():
            self._transaction = None
            self._started = False
    
        if not isinstance(error, AsyncAdapt_asyncpg_dbapi.Error):
            exception_mapping = self.dbapi._asyncpg_error_translate
    
            for super_ in type(error).__mro__:
                if super_ in exception_mapping:
                    translated_error = exception_mapping[super_](
                        &quot;%s: %s&quot; % (type(error), error)
                    )
                    translated_error.pgcode = (
                        translated_error.sqlstate
                    ) = getattr(error, &quot;sqlstate&quot;, None)
&gt;                   raise translated_error from error
E                   sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) &lt;class 'asyncpg.exceptions.UniqueViolationError'&gt;: duplicate key value violates unique constraint &quot;license_setting_external_id_key&quot;
E                   DETAIL:  Key (external_id)=(c2eed5e5-e234-48a8-8ee1-ee110a8dd980) already exists.
E                   [SQL: INSERT INTO license_setting (external_id, reassign_licenses_on_license_pool_expiration, allowed_reassigned_percentage, account_id) VALUES (%s, %s, %s, %s) RETURNING license_setting.id]
E                   [parameters: (UUID('c2eed5e5-e234-48a8-8ee1-ee110a8dd980'), True, 10, UUID('79644b11-b054-4dc2-927b-cf76c9da89e8'))]
E                   (Background on this error at: https://sqlalche.me/e/14/gkpj)"><pre class="notranslate"><code class="notranslate">self = &lt;sqlalchemy.dialects.postgresql.asyncpg.AsyncAdapt_asyncpg_connection object at 0x10a30be40&gt;
error = UniqueViolationError('duplicate key value violates unique constraint "license_setting_external_id_key"')

    def _handle_exception(self, error):
        if self._connection.is_closed():
            self._transaction = None
            self._started = False
    
        if not isinstance(error, AsyncAdapt_asyncpg_dbapi.Error):
            exception_mapping = self.dbapi._asyncpg_error_translate
    
            for super_ in type(error).__mro__:
                if super_ in exception_mapping:
                    translated_error = exception_mapping[super_](
                        "%s: %s" % (type(error), error)
                    )
                    translated_error.pgcode = (
                        translated_error.sqlstate
                    ) = getattr(error, "sqlstate", None)
&gt;                   raise translated_error from error
E                   sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) &lt;class 'asyncpg.exceptions.UniqueViolationError'&gt;: duplicate key value violates unique constraint "license_setting_external_id_key"
E                   DETAIL:  Key (external_id)=(c2eed5e5-e234-48a8-8ee1-ee110a8dd980) already exists.
E                   [SQL: INSERT INTO license_setting (external_id, reassign_licenses_on_license_pool_expiration, allowed_reassigned_percentage, account_id) VALUES (%s, %s, %s, %s) RETURNING license_setting.id]
E                   [parameters: (UUID('c2eed5e5-e234-48a8-8ee1-ee110a8dd980'), True, 10, UUID('79644b11-b054-4dc2-927b-cf76c9da89e8'))]
E                   (Background on this error at: https://sqlalche.me/e/14/gkpj)
</code></pre></div>
<p dir="auto">Am I missing something? Is it the intended behavior?</p>
<h3 dir="auto">To Reproduce</h3>
<p dir="auto">Here the test I'm doing:</p>
<div class="highlight highlight-source-python notranslate position-relative overflow-auto" dir="auto" data-snippet-clipboard-copy-content=" @pytest.mark.asyncio
    async def test_that_retrieved_license_setting_is_the_same_of_the_added_entity(self, session: AsyncSession) -&gt; None:
        # Setup
        license_setting_one = LicenseSetting(
            account_id=uuid.uuid4(),
            reassign_licenses_on_license_pool_expiration=True,
            allowed_reassigned_percentage=10,
        )

        license_setting_two = LicenseSetting(
            id=license_setting_one.id, # same primary key of license_setting_one -&gt; We are the same entity!
            account_id=uuid.uuid4(),
            reassign_licenses_on_license_pool_expiration=True,
            allowed_reassigned_percentage=10,
        )

        # SUT
        session.begin_nested()
        session.add(license_setting_one)
        await session.commit()
        session.expunge(license_setting_one)

        session.begin_nested()
        session.add(license_setting_two)
        await session.commit()
        session.expunge(license_setting_two)

        query = select(LicenseSetting).limit(1)
        result = await session.execute(query)

        retrieved_license_setting: LicenseSetting = result.scalar()

        # Assert
        assert retrieved_license_setting is not None
        assert retrieved_license_setting == license_setting_two
"><pre class="notranslate"> <span class="pl-en">@<span class="pl-s1">pytest</span>.<span class="pl-s1">mark</span>.<span class="pl-s1">asyncio</span></span>
    <span class="pl-k">async</span> <span class="pl-k">def</span> <span class="pl-en">test_that_retrieved_license_setting_is_the_same_of_the_added_entity</span>(<span class="pl-s1">self</span>, <span class="pl-s1">session</span>: <span class="pl-v">AsyncSession</span>) <span class="pl-c1">-&gt;</span> <span class="pl-c1">None</span>:
        <span class="pl-c"># Setup</span>
        <span class="pl-s1">license_setting_one</span> <span class="pl-c1">=</span> <span class="pl-v">LicenseSetting</span>(
            <span class="pl-s1">account_id</span><span class="pl-c1">=</span><span class="pl-s1">uuid</span>.<span class="pl-en">uuid4</span>(),
            <span class="pl-s1">reassign_licenses_on_license_pool_expiration</span><span class="pl-c1">=</span><span class="pl-c1">True</span>,
            <span class="pl-s1">allowed_reassigned_percentage</span><span class="pl-c1">=</span><span class="pl-c1">10</span>,
        )

        <span class="pl-s1">license_setting_two</span> <span class="pl-c1">=</span> <span class="pl-v">LicenseSetting</span>(
            <span class="pl-s1">id</span><span class="pl-c1">=</span><span class="pl-s1">license_setting_one</span>.<span class="pl-s1">id</span>, <span class="pl-c"># same primary key of license_setting_one -&gt; We are the same entity!</span>
            <span class="pl-s1">account_id</span><span class="pl-c1">=</span><span class="pl-s1">uuid</span>.<span class="pl-en">uuid4</span>(),
            <span class="pl-s1">reassign_licenses_on_license_pool_expiration</span><span class="pl-c1">=</span><span class="pl-c1">True</span>,
            <span class="pl-s1">allowed_reassigned_percentage</span><span class="pl-c1">=</span><span class="pl-c1">10</span>,
        )

        <span class="pl-c"># SUT</span>
        <span class="pl-s1">session</span>.<span class="pl-en">begin_nested</span>()
        <span class="pl-s1">session</span>.<span class="pl-en">add</span>(<span class="pl-s1">license_setting_one</span>)
        <span class="pl-k">await</span> <span class="pl-s1">session</span>.<span class="pl-en">commit</span>()
        <span class="pl-s1">session</span>.<span class="pl-en">expunge</span>(<span class="pl-s1">license_setting_one</span>)

        <span class="pl-s1">session</span>.<span class="pl-en">begin_nested</span>()
        <span class="pl-s1">session</span>.<span class="pl-en">add</span>(<span class="pl-s1">license_setting_two</span>)
        <span class="pl-k">await</span> <span class="pl-s1">session</span>.<span class="pl-en">commit</span>()
        <span class="pl-s1">session</span>.<span class="pl-en">expunge</span>(<span class="pl-s1">license_setting_two</span>)

        <span class="pl-s1">query</span> <span class="pl-c1">=</span> <span class="pl-en">select</span>(<span class="pl-v">LicenseSetting</span>).<span class="pl-en">limit</span>(<span class="pl-c1">1</span>)
        <span class="pl-s1">result</span> <span class="pl-c1">=</span> <span class="pl-k">await</span> <span class="pl-s1">session</span>.<span class="pl-en">execute</span>(<span class="pl-s1">query</span>)

        <span class="pl-s1">retrieved_license_setting</span>: <span class="pl-v">LicenseSetting</span> <span class="pl-c1">=</span> <span class="pl-s1">result</span>.<span class="pl-en">scalar</span>()

        <span class="pl-c"># Assert</span>
        <span class="pl-k">assert</span> <span class="pl-s1">retrieved_license_setting</span> <span class="pl-c1">is</span> <span class="pl-c1">not</span> <span class="pl-c1">None</span>
        <span class="pl-k">assert</span> <span class="pl-s1">retrieved_license_setting</span> <span class="pl-c1">==</span> <span class="pl-s1">license_setting_two</span></pre></div>
<h3 dir="auto">Error</h3>
<div class="snippet-clipboard-content notranslate position-relative overflow-auto" data-snippet-clipboard-copy-content="self = &lt;sqlalchemy.dialects.postgresql.asyncpg.AsyncAdapt_asyncpg_connection object at 0x10a30be40&gt;
error = UniqueViolationError('duplicate key value violates unique constraint &quot;license_setting_external_id_key&quot;')

    def _handle_exception(self, error):
        if self._connection.is_closed():
            self._transaction = None
            self._started = False
    
        if not isinstance(error, AsyncAdapt_asyncpg_dbapi.Error):
            exception_mapping = self.dbapi._asyncpg_error_translate
    
            for super_ in type(error).__mro__:
                if super_ in exception_mapping:
                    translated_error = exception_mapping[super_](
                        &quot;%s: %s&quot; % (type(error), error)
                    )
                    translated_error.pgcode = (
                        translated_error.sqlstate
                    ) = getattr(error, &quot;sqlstate&quot;, None)
&gt;                   raise translated_error from error
E                   sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) &lt;class 'asyncpg.exceptions.UniqueViolationError'&gt;: duplicate key value violates unique constraint &quot;license_setting_external_id_key&quot;
E                   DETAIL:  Key (external_id)=(c2eed5e5-e234-48a8-8ee1-ee110a8dd980) already exists.
E                   [SQL: INSERT INTO license_setting (external_id, reassign_licenses_on_license_pool_expiration, allowed_reassigned_percentage, account_id) VALUES (%s, %s, %s, %s) RETURNING license_setting.id]
E                   [parameters: (UUID('c2eed5e5-e234-48a8-8ee1-ee110a8dd980'), True, 10, UUID('79644b11-b054-4dc2-927b-cf76c9da89e8'))]
E                   (Background on this error at: https://sqlalche.me/e/14/gkpj)
"><pre class="notranslate"><code class="notranslate">self = &lt;sqlalchemy.dialects.postgresql.asyncpg.AsyncAdapt_asyncpg_connection object at 0x10a30be40&gt;
error = UniqueViolationError('duplicate key value violates unique constraint "license_setting_external_id_key"')

    def _handle_exception(self, error):
        if self._connection.is_closed():
            self._transaction = None
            self._started = False
    
        if not isinstance(error, AsyncAdapt_asyncpg_dbapi.Error):
            exception_mapping = self.dbapi._asyncpg_error_translate
    
            for super_ in type(error).__mro__:
                if super_ in exception_mapping:
                    translated_error = exception_mapping[super_](
                        "%s: %s" % (type(error), error)
                    )
                    translated_error.pgcode = (
                        translated_error.sqlstate
                    ) = getattr(error, "sqlstate", None)
&gt;                   raise translated_error from error
E                   sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) &lt;class 'asyncpg.exceptions.UniqueViolationError'&gt;: duplicate key value violates unique constraint "license_setting_external_id_key"
E                   DETAIL:  Key (external_id)=(c2eed5e5-e234-48a8-8ee1-ee110a8dd980) already exists.
E                   [SQL: INSERT INTO license_setting (external_id, reassign_licenses_on_license_pool_expiration, allowed_reassigned_percentage, account_id) VALUES (%s, %s, %s, %s) RETURNING license_setting.id]
E                   [parameters: (UUID('c2eed5e5-e234-48a8-8ee1-ee110a8dd980'), True, 10, UUID('79644b11-b054-4dc2-927b-cf76c9da89e8'))]
E                   (Background on this error at: https://sqlalche.me/e/14/gkpj)

</code></pre></div>
<h3 dir="auto">Versions</h3>
<ul dir="auto">
<li>OS: MacOs Catalina 10.15.7</li>
<li>Python: 3.9</li>
<li>SQLAlchemy: 1.4</li>
<li>Database: Postgres</li>
<li>DBAPI: 2.0 (async engine)</li>
</ul>
<h3 dir="auto">Additional context</h3>
<p dir="auto"><em>No response</em></p>
<h3 dir="auto">Describe the use case</h3>
<p dir="auto">Accroding to<br>
<a href="https://mysqlclient.readthedocs.io/user_guide.html" rel="nofollow">https://mysqlclient.readthedocs.io/user_guide.html</a><br>
The mysqldb api provide a method to retrieve the warning count of the db.</p>
<h3 dir="auto">Databases / Backends / Drivers targeted</h3>
<p dir="auto">It is for mysql database.</p>
<h3 dir="auto">Example Use</h3>
<p dir="auto">These is particularly usful when using <code class="notranslate">insert ignore .....</code>  to get how many duplicates key you encounter.</p>
<p dir="auto">Example use:<br>
<code class="notranslate">result = engine.execute('insert ignore .....') </code><br>
<code class="notranslate">duplicate_count = result.warning_count()</code></p>
<h3 dir="auto">Additional context</h3>
<p dir="auto"><em>No response</em></p>