MCPcopy
hub / github.com/sqlalchemy/sqlalchemy / _decorate_polymorphic_switch

Function _decorate_polymorphic_switch

lib/sqlalchemy/orm/loading.py:1447–1533  ·  view source on GitHub ↗
(
    instance_fn,
    context,
    query_entity,
    mapper,
    result,
    path,
    polymorphic_discriminator,
    adapter,
    ensure_no_pk,
)

Source from the content-addressed store, hash-verified

1445
1446
1447def _decorate_polymorphic_switch(
1448 instance_fn,
1449 context,
1450 query_entity,
1451 mapper,
1452 result,
1453 path,
1454 polymorphic_discriminator,
1455 adapter,
1456 ensure_no_pk,
1457):
1458 if polymorphic_discriminator is not None:
1459 polymorphic_on = polymorphic_discriminator
1460 else:
1461 polymorphic_on = mapper.polymorphic_on
1462 if polymorphic_on is None:
1463 return instance_fn
1464
1465 if adapter:
1466 polymorphic_on = adapter.columns[polymorphic_on]
1467
1468 def configure_subclass_mapper(discriminator):
1469 try:
1470 sub_mapper = mapper.polymorphic_map[discriminator]
1471 except KeyError:
1472 raise AssertionError(
1473 "No such polymorphic_identity %r is defined" % discriminator
1474 )
1475 else:
1476 if sub_mapper is mapper:
1477 return None
1478 elif not sub_mapper.isa(mapper):
1479 return False
1480
1481 return _instance_processor(
1482 query_entity,
1483 sub_mapper,
1484 context,
1485 result,
1486 path,
1487 adapter,
1488 _polymorphic_from=mapper,
1489 )
1490
1491 polymorphic_instances = util.PopulateDict(configure_subclass_mapper)
1492
1493 getter = result._getter(polymorphic_on)
1494
1495 def polymorphic_instance(row):
1496 discriminator = getter(row)
1497 if discriminator is not None:
1498 _instance = polymorphic_instances[discriminator]
1499 if _instance:
1500 return _instance(row)
1501 elif _instance is False:
1502 identitykey = ensure_no_pk(row)
1503
1504 if identitykey:

Callers 1

_instance_processorFunction · 0.85

Calls 1

_getterMethod · 0.45

Tested by

no test coverage detected