Construct schema information via engine-specific queries of the tables in the DB. :return: Schema information of the following form, sorted by db_table_name. [ ("db_table_name", [ ("db_
(db_connection)
| 76 | |
| 77 | |
| 78 | def build_schema_info(db_connection): |
| 79 | """ |
| 80 | Construct schema information via engine-specific queries of the |
| 81 | tables in the DB. |
| 82 | |
| 83 | :return: Schema information of the following form, |
| 84 | sorted by db_table_name. |
| 85 | [ |
| 86 | ("db_table_name", |
| 87 | [ |
| 88 | ("db_column_name", "DbFieldType"), |
| 89 | (...), |
| 90 | ] |
| 91 | ) |
| 92 | ] |
| 93 | |
| 94 | """ |
| 95 | connection = db_connection.as_django_connection() |
| 96 | ret = [] |
| 97 | with connection.cursor() as cursor: |
| 98 | tables_to_introspect = connection.introspection.table_names( |
| 99 | cursor, include_views=_include_views() |
| 100 | ) |
| 101 | |
| 102 | for table_name in tables_to_introspect: |
| 103 | if not _include_table(table_name): |
| 104 | continue |
| 105 | td = [] |
| 106 | table_description = connection.introspection.get_table_description( |
| 107 | cursor, table_name |
| 108 | ) |
| 109 | for row in table_description: |
| 110 | column_name = row[0] |
| 111 | try: |
| 112 | field_type = connection.introspection.get_field_type( |
| 113 | row[1], row |
| 114 | ) |
| 115 | except KeyError: |
| 116 | field_type = "Unknown" |
| 117 | td.append((column_name, field_type)) |
| 118 | ret.append((table_name, td)) |
| 119 | return ret |
| 120 | |
| 121 |
no test coverage detected