def apply_foreign_keys()

in src/databao_context_engine/plugins/databases/table_builder.py [0:0]

• 22 lines of code
• 6 McCabe index (conditional complexity)

    def apply_foreign_keys(self, fk_cols: list[dict] | None) -> None:
        fk_groups = group_rows(fk_cols, ("table_name", "constraint_name"))
        by_table: dict[str, list[ForeignKey]] = defaultdict(list)

        for (table_name, cname), grp in fk_groups.items():
            grp.sort(key=lambda r: sort_position_by_key(r, "position"))
            first = grp[0]

            ref_schema = first.get("ref_schema")
            ref_table = first.get("ref_table")
            referenced = f"{ref_schema}.{ref_table}" if ref_schema and ref_table else ""

            by_table[table_name].append(
                ForeignKey(
                    name=cname,
                    mapping=[ForeignKeyColumnMap(from_column=r["from_column"], to_column=r["to_column"]) for r in grp],
                    referenced_table=referenced,
                    on_update=first.get("on_update"),
                    on_delete=first.get("on_delete"),
                    enforced=coerce_bool(first.get("enforced"), default=True),
                    validated=coerce_bool(first.get("validated"), default=True),
                )
            )

        for table_name, fks in by_table.items():
            self.get_or_create_table(table_name).foreign_keys = fks