(meta: sa.MetaData, min_cols, max_cols, dialect_name)
| 15 | |
| 16 | |
| 17 | def generate_table(meta: sa.MetaData, min_cols, max_cols, dialect_name): |
| 18 | col_number = random.randint(min_cols, max_cols) |
| 19 | table_num = len(meta.tables) |
| 20 | add_identity = random.random() > 0.90 |
| 21 | identity = sa.Identity( |
| 22 | always=random.randint(0, 1), |
| 23 | start=random.randint(1, 100), |
| 24 | increment=random.randint(1, 7), |
| 25 | ) |
| 26 | is_mssql = dialect_name == "mssql" |
| 27 | cols = [] |
| 28 | for i in range(col_number - (0 if is_mssql else add_identity)): |
| 29 | args = [] |
| 30 | if random.random() < 0.99 or table_num == 0: |
| 31 | if is_mssql and add_identity and i == 0: |
| 32 | args.append(sa.Integer) |
| 33 | args.append(identity) |
| 34 | else: |
| 35 | args.append(random.choice(types)) |
| 36 | else: |
| 37 | target = random.randint(0, table_num - 1) |
| 38 | args.append(sa.ForeignKey(f"table_{target}.table_{target}_col_1")) |
| 39 | cols.append( |
| 40 | sa.Column( |
| 41 | f"table_{table_num}_col_{i + 1}", |
| 42 | *args, |
| 43 | primary_key=i == 0, |
| 44 | comment=( |
| 45 | f"primary key of table_{table_num}" if i == 0 else None |
| 46 | ), |
| 47 | index=random.random() > 0.97 and i > 0, |
| 48 | unique=random.random() > 0.97 and i > 0, |
| 49 | ) |
| 50 | ) |
| 51 | if add_identity and not is_mssql: |
| 52 | cols.append( |
| 53 | sa.Column( |
| 54 | f"table_{table_num}_col_{col_number}", |
| 55 | sa.Integer, |
| 56 | identity, |
| 57 | ) |
| 58 | ) |
| 59 | args = () |
| 60 | if table_num % 3 == 0: |
| 61 | # mysql can't do check constraint on PK col |
| 62 | args = (sa.CheckConstraint(cols[1].is_not(None)),) |
| 63 | return sa.Table( |
| 64 | f"table_{table_num}", |
| 65 | meta, |
| 66 | *cols, |
| 67 | *args, |
| 68 | comment=f"comment for table_{table_num}" if table_num % 2 else None, |
| 69 | ) |
| 70 | |
| 71 | |
| 72 | def generate_meta(schema_name, table_number, min_cols, max_cols, dialect_name): |
no test coverage detected