MCPcopy Create free account
hub / github.com/sqlalchemy/sqlalchemy / _deep_annotate

Function _deep_annotate

lib/sqlalchemy/sql/annotation.py:410–479  ·  view source on GitHub ↗

Deep copy the given ClauseElement, annotating each element with the given annotations dictionary. Elements within the exclude collection will be cloned but not annotated.

(
    element: _SA,
    annotations: _AnnotationDict,
    exclude: Optional[Sequence[SupportsAnnotations]] = None,
    *,
    detect_subquery_cols: bool = False,
    ind_cols_on_fromclause: bool = False,
    annotate_callable: Optional[
        Callable[[SupportsAnnotations, _AnnotationDict], SupportsAnnotations]
    ] = None,
)

Source from the content-addressed store, hash-verified

408
409
410def _deep_annotate(
411 element: _SA,
412 annotations: _AnnotationDict,
413 exclude: Optional[Sequence[SupportsAnnotations]] = None,
414 *,
415 detect_subquery_cols: bool = False,
416 ind_cols_on_fromclause: bool = False,
417 annotate_callable: Optional[
418 Callable[[SupportsAnnotations, _AnnotationDict], SupportsAnnotations]
419 ] = None,
420) -> _SA:
421 """Deep copy the given ClauseElement, annotating each element
422 with the given annotations dictionary.
423
424 Elements within the exclude collection will be cloned but not annotated.
425
426 """
427
428 # annotated objects hack the __hash__() method so if we want to
429 # uniquely process them we have to use id()
430
431 cloned_ids: Dict[int, SupportsAnnotations] = {}
432
433 def clone(elem: SupportsAnnotations, **kw: Any) -> SupportsAnnotations:
434 # ind_cols_on_fromclause means make sure an AnnotatedFromClause
435 # has its own .c collection independent of that which its proxying.
436 # this is used specifically by orm.LoaderCriteriaOption to break
437 # a reference cycle that it's otherwise prone to building,
438 # see test_relationship_criteria->
439 # test_loader_criteria_subquery_w_same_entity. logic here was
440 # changed for #8796 and made explicit; previously it occurred
441 # by accident
442
443 kw["detect_subquery_cols"] = detect_subquery_cols
444 id_ = id(elem)
445
446 if id_ in cloned_ids:
447 return cloned_ids[id_]
448
449 if (
450 exclude
451 and hasattr(elem, "proxy_set")
452 and elem.proxy_set.intersection(exclude)
453 ):
454 newelem = elem._clone(clone=clone, **kw)
455 elif annotations != elem._annotations:
456 if detect_subquery_cols and elem._is_immutable:
457 to_annotate = elem._clone(clone=clone, **kw)
458 else:
459 to_annotate = elem
460 if annotate_callable:
461 newelem = annotate_callable(to_annotate, annotations)
462 else:
463 newelem = _safe_annotate(to_annotate, annotations)
464 else:
465 newelem = elem
466
467 newelem._copy_internals(

Callers 2

join_targetsMethod · 0.85
test_annotated_fetchMethod · 0.85

Calls 2

castFunction · 0.85
cloneFunction · 0.70

Tested by 1

test_annotated_fetchMethod · 0.68