Issue 1:
Describe the bug

Hi,

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.

I want that from a codebase point of view the UUID is the primary key, not the integer field.

@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["external_id"]], # here I specify I want use the field external_id as primary key
exclude_properties=["id"], # real 'id' of the table is not mapped to my domain entity
properties={
'id': license_setting_table.c["external_id"] # here I specify I want to map db 'external_id' to entity 'id'
})

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.

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.

The error I receive:

self = <sqlalchemy.dialects.postgresql.asyncpg.AsyncAdapt_asyncpg_connection object at 0x10a30be40>
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)
> raise translated_error from error
E sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) <class 'asyncpg.exceptions.UniqueViolationError'>: 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)

Am I missing something? Is it the intended behavior?

To Reproduce

Here the test I'm doing:

@pytest.mark.asyncio
async def test_that_retrieved_license_setting_is_the_same_of_the_added_entity(self, session: AsyncSession) -> 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 -> 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

Error

self = <sqlalchemy.dialects.postgresql.asyncpg.AsyncAdapt_asyncpg_connection object at 0x10a30be40>
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)
> raise translated_error from error
E sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) <class 'asyncpg.exceptions.UniqueViolationError'>: 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)

Versions

OS: MacOs Catalina 10.15.7

Python: 3.9

SQLAlchemy: 1.4

Database: Postgres

DBAPI: 2.0 (async engine)

Additional context

No response

Issue 2:
Describe the use case

Accroding to

https://mysqlclient.readthedocs.io/user_guide.html

The mysqldb api provide a method to retrieve the warning count of the db.

Databases / Backends / Drivers targeted

It is for mysql database.

Example Use

These is particularly usful when using insert ignore ..... to get how many duplicates key you encounter.

Example use:

result = engine.execute('insert ignore .....')

duplicate_count = result.warning_count()

Additional context

No response