(server,
db_name,
table_name,
constraint_type="unique",
constraint_name="test_unique")
| 325 | |
| 326 | |
| 327 | def create_constraint(server, |
| 328 | db_name, |
| 329 | table_name, |
| 330 | constraint_type="unique", |
| 331 | constraint_name="test_unique"): |
| 332 | try: |
| 333 | connection = get_db_connection( |
| 334 | db_name, |
| 335 | server['username'], |
| 336 | server['db_password'], |
| 337 | server['host'], |
| 338 | server['port'], |
| 339 | server['sslmode'] |
| 340 | ) |
| 341 | old_isolation_level = connection.isolation_level |
| 342 | set_isolation_level(connection, 0) |
| 343 | pg_cursor = connection.cursor() |
| 344 | pg_cursor.execute(''' |
| 345 | ALTER TABLE "%s" |
| 346 | ADD CONSTRAINT "%s" %s (some_column) |
| 347 | ''' % (table_name, constraint_name, constraint_type.upper())) |
| 348 | |
| 349 | set_isolation_level(connection, old_isolation_level) |
| 350 | connection.commit() |
| 351 | |
| 352 | except Exception: |
| 353 | traceback.print_exc(file=sys.stderr) |
| 354 | |
| 355 | |
| 356 | def create_type(server, db_name, type_name, type_fields=[]): |
nothing calls this directly
no test coverage detected