Resolves a nested field string like events__participants__name and returns the pypika term, required joins and the Field that can be used for converting the value.
(
model: type[Model], table: Table, field: str
)
| 104 | |
| 105 | |
| 106 | def resolve_nested_field( |
| 107 | model: type[Model], table: Table, field: str |
| 108 | ) -> tuple[Term, list[TableCriterionTuple], Field | None]: |
| 109 | """ |
| 110 | Resolves a nested field string like events__participants__name and |
| 111 | returns the pypika term, required joins and the Field that can be used for |
| 112 | converting the value. |
| 113 | """ |
| 114 | joins = [] |
| 115 | fields = expand_lookup_expression(model, field) |
| 116 | |
| 117 | for iter_field in fields[:-1]: |
| 118 | related_field = cast(RelationalField, iter_field) |
| 119 | joins.extend(get_joins_for_related_field(table, related_field, iter_field.model_field_name)) |
| 120 | |
| 121 | model = related_field.related_model |
| 122 | related_table: Table = related_field.related_model._meta.basetable |
| 123 | if isinstance(related_field, (ForeignKeyFieldInstance, ManyToManyFieldInstance)): |
| 124 | related_table = related_table.as_( |
| 125 | f"{table.get_table_name()}__{iter_field.model_field_name}" |
| 126 | ) |
| 127 | table = related_table |
| 128 | |
| 129 | last_field = fields[-1] |
| 130 | if last_field.model_field_name in model._meta.fetch_fields: |
| 131 | related_field = cast(RelationalField, last_field) |
| 132 | related_field_meta = related_field.related_model._meta |
| 133 | |
| 134 | joins.extend( |
| 135 | get_joins_for_related_field(table, related_field, related_field.model_field_name) |
| 136 | ) |
| 137 | related_table = related_field_meta.basetable |
| 138 | |
| 139 | if isinstance(related_field, BackwardFKRelation): |
| 140 | if table == related_table: |
| 141 | related_table = related_table.as_( |
| 142 | f"{table.get_table_name()}__{related_field.model_field_name}" |
| 143 | ) |
| 144 | elif isinstance(related_field, ManyToManyFieldInstance): |
| 145 | related_table = related_table.as_( |
| 146 | f"{table.get_table_name()}__{related_field.model_field_name}" |
| 147 | ) |
| 148 | |
| 149 | term = related_table[related_field_meta.db_pk_column] |
| 150 | else: |
| 151 | if last_field.source_field: |
| 152 | term = table[last_field.source_field] |
| 153 | else: |
| 154 | term = table[last_field.model_field_name] |
| 155 | |
| 156 | if last_field: # pragma: nobranch |
| 157 | func = last_field.get_for_dialect(model._meta.db.capabilities.dialect, "function_cast") |
| 158 | if func: |
| 159 | term = func(last_field, term) |
| 160 | |
| 161 | return term, joins, last_field |
| 162 | |
| 163 |
no test coverage detected
searching dependent graphs…