(factory, args, kw)
| 47 | |
| 48 | |
| 49 | def _schema_column(factory, args, kw): |
| 50 | test_opts = {k: kw.pop(k) for k in list(kw) if k.startswith("test_")} |
| 51 | |
| 52 | if not config.requirements.foreign_key_ddl.enabled_for_config(config): |
| 53 | args = [arg for arg in args if not isinstance(arg, schema.ForeignKey)] |
| 54 | |
| 55 | construct = factory(*args, **kw) |
| 56 | |
| 57 | if factory is schema.Column: |
| 58 | col = construct |
| 59 | else: |
| 60 | col = construct.column |
| 61 | |
| 62 | if test_opts.get("test_needs_autoincrement", False) and kw.get( |
| 63 | "primary_key", False |
| 64 | ): |
| 65 | if col.default is None and col.server_default is None: |
| 66 | col.autoincrement = True |
| 67 | |
| 68 | # allow any test suite to pick up on this |
| 69 | col.info["test_needs_autoincrement"] = True |
| 70 | |
| 71 | # hardcoded rule for oracle; this should |
| 72 | # be moved out |
| 73 | if exclusions.against(config._current, "oracle"): |
| 74 | |
| 75 | def add_seq(c, tbl): |
| 76 | c._init_items( |
| 77 | schema.Sequence( |
| 78 | _truncate_name( |
| 79 | config.db.dialect, tbl.name + "_" + c.name + "_seq" |
| 80 | ), |
| 81 | optional=True, |
| 82 | ) |
| 83 | ) |
| 84 | |
| 85 | event.listen(col, "after_parent_attach", add_seq, propagate=True) |
| 86 | return construct |
| 87 | |
| 88 | |
| 89 | class eq_type_affinity: |
no test coverage detected