(
self,
_reflect_info: _ReflectionInfo,
table_key: TableKey,
table: sa_schema.Table,
cols_by_orig_name: Dict[str, sa_schema.Column[Any]],
exclude_columns: Collection[str],
)
| 1700 | table.append_column(col, replace_existing=True) |
| 1701 | |
| 1702 | def _reflect_pk( |
| 1703 | self, |
| 1704 | _reflect_info: _ReflectionInfo, |
| 1705 | table_key: TableKey, |
| 1706 | table: sa_schema.Table, |
| 1707 | cols_by_orig_name: Dict[str, sa_schema.Column[Any]], |
| 1708 | exclude_columns: Collection[str], |
| 1709 | ) -> None: |
| 1710 | pk_cons = _reflect_info.pk_constraint.get(table_key) |
| 1711 | if pk_cons: |
| 1712 | pk_cols = [ |
| 1713 | cols_by_orig_name[pk] |
| 1714 | for pk in pk_cons["constrained_columns"] |
| 1715 | if pk in cols_by_orig_name and pk not in exclude_columns |
| 1716 | ] |
| 1717 | |
| 1718 | # update pk constraint name, comment and dialect_kwargs |
| 1719 | table.primary_key.name = pk_cons.get("name") |
| 1720 | table.primary_key.comment = pk_cons.get("comment", None) |
| 1721 | dialect_options = pk_cons.get("dialect_options") |
| 1722 | if dialect_options: |
| 1723 | table.primary_key.dialect_kwargs.update(dialect_options) |
| 1724 | |
| 1725 | # tell the PKConstraint to re-initialize |
| 1726 | # its column collection |
| 1727 | table.primary_key._reload(pk_cols) |
| 1728 | |
| 1729 | def _reflect_fk( |
| 1730 | self, |
no test coverage detected