(cls, class_type, primary_key, indexes=None, owner_column=None, owner_type=None)
| 20 | |
| 21 | @classmethod |
| 22 | def load_schema(cls, class_type, primary_key, indexes=None, owner_column=None, owner_type=None): |
| 23 | # type: (Type, Sequence[str], Optional[Dict[str, Union[str, Sequence[str]]]], Optional[str], Optional[Type]) -> TableSchema |
| 24 | schema = cls() |
| 25 | schema.class_type = class_type |
| 26 | schema.table_name = class_type.__name__ |
| 27 | obj_values = vars(class_type()) |
| 28 | for field_name, value in obj_values.items(): |
| 29 | column_name = field_name.lower() |
| 30 | field_type = None # type: Optional[Type] |
| 31 | if isinstance(value, str): |
| 32 | field_type = str |
| 33 | elif isinstance(value, bool): |
| 34 | field_type = bool |
| 35 | elif isinstance(value, int): |
| 36 | field_type = int |
| 37 | elif isinstance(value, float): |
| 38 | field_type = float |
| 39 | elif isinstance(value, (bytes, bytearray)): |
| 40 | field_type = bytes |
| 41 | elif value is None: |
| 42 | logging.debug( |
| 43 | 'load_schema: Attribute \"%s\" in class \"%s\" is skipped since it is None', |
| 44 | field_name, schema.table_name) |
| 45 | else: |
| 46 | logging.debug( |
| 47 | 'load_schema: Unsupported type for attribute \"%s\" in class \"%s\". Skipping', |
| 48 | field_name, schema.table_name) |
| 49 | if field_type: |
| 50 | schema.class_fields[column_name] = FieldSchema(field_name, field_type) |
| 51 | schema.columns.append(field_name) |
| 52 | |
| 53 | if isinstance(primary_key, str): |
| 54 | schema.primary_key.append(primary_key) |
| 55 | elif isinstance(primary_key, (list, set)): |
| 56 | schema.primary_key.extend(primary_key) |
| 57 | else: |
| 58 | if not owner_column: |
| 59 | raise ValueError(f'Schema \"{schema.table_name}\" does not have either primary key or owner column') |
| 60 | if len(schema.columns) == 0: |
| 61 | raise ValueError(f'Table {schema.table_name} does not have primary key defined') |
| 62 | |
| 63 | for column in schema.primary_key: |
| 64 | if column.lower() not in schema.class_fields: |
| 65 | raise ValueError(f'Primary Key: Table {schema.table_name} does not have column {column}') |
| 66 | |
| 67 | if isinstance(indexes, dict): |
| 68 | schema.indexes = {} |
| 69 | for index_name, index_columns in indexes.items(): |
| 70 | if isinstance(index_columns, str): |
| 71 | index_columns = [index_columns] |
| 72 | if not isinstance(index_columns, (list, tuple)): |
| 73 | raise ValueError(f'Index \"{index_name}\": invalid columns') |
| 74 | for column in index_columns: |
| 75 | if column.lower() not in schema.class_fields: |
| 76 | raise ValueError( |
| 77 | f'Index \"{index_name}\": Table {schema.table_name} does not have column {column}') |
| 78 | schema.indexes[index_name] = [x for x in index_columns] |
| 79 |
no test coverage detected