193 ) -> list[sqlalchemy.schema.SchemaItem]:
194 """Return set of constraints/indices in a table.
195
196 Parameters
197 ----------
198 table : `schema_model.Table`
199 Table model.
200 table_map : `~collections.abc.Mapping`
201 MApping of table ID to sqlalchemy table definition for tables
202 that already exist, this must include all tables referenced by
203 foreign keys in ``table``.
204
205 Returns
206 -------
207 constraints : `list` [`sqlalchemy.schema.SchemaItem`]
208 List of SQLAlchemy index/constraint objects.
209 """
210 constraints: list[sqlalchemy.schema.SchemaItem] = []
211 if table.primary_key:
212
213 name = self._prefix + table.name + "_pk"
214 constraints.append(
215 sqlalchemy.schema.PrimaryKeyConstraint(*[column.name for column in table.primary_key])
216 )
217 for index in table.indexes:
218 if index.expressions:
219 raise TypeError(f"Expression indices are not supported: {table}")
220 name = self._prefix + index.name if index.name else ""
221 constraints.append(sqlalchemy.schema.Index(name, *[column.name for column in index.columns]))
222 for constraint in table.constraints:
223 constr_name: str | None = None
224 if constraint.name:
225 constr_name = self._prefix + constraint.name
226 if isinstance(constraint, schema_model.UniqueConstraint):
227 constraints.append(
228 sqlalchemy.schema.UniqueConstraint(
229 *[column.name for column in constraint.columns], name=constr_name
230 )
231 )
232 elif isinstance(constraint, schema_model.ForeignKeyConstraint):
233 column_names = [col.name for col in constraint.columns]
234 foreign_table = table_map[constraint.referenced_table.id]
235 refcolumns = [foreign_table.columns[col.name] for col in constraint.referenced_columns]
236 constraints.append(
237 sqlalchemy.schema.ForeignKeyConstraint(
238 columns=column_names,
239 refcolumns=refcolumns,
240 name=constr_name,
241 deferrable=constraint.deferrable,
242 initially=constraint.initially,
243 onupdate=constraint.onupdate,
244 ondelete=constraint.ondelete,
245 )
246 )
247 elif isinstance(constraint, schema_model.CheckConstraint):
248 constraints.append(
249 sqlalchemy.schema.CheckConstraint(
250 constraint.expression,
251 name=constr_name,
252 deferrable=constraint.deferrable,
253 initially=constraint.initially,
254 )
255 )
256 else:
257 raise TypeError(f"Unknown constraint type: {constraint}")
258
259 return constraints
260