Builds initialization SQL for MySQL.
(entry: dict)
| 252 | |
| 253 | @staticmethod |
| 254 | def _build_init_sql(entry: dict) -> List[Union[str, Tuple[str, Sequence[str]]]]: |
| 255 | """Builds initialization SQL for MySQL.""" |
| 256 | tables = entry["table"] if isinstance(entry["table"], list) else [entry["table"]] |
| 257 | final_sql = [] |
| 258 | for table in tables: |
| 259 | name = table["table_name"] |
| 260 | columns = ",".join([f"`{c['name']}` TEXT" for c in table["table_info"]["columns"]]) |
| 261 | column_names = ",".join([f"`{c['name']}`" for c in table["table_info"]["columns"]]) |
| 262 | items = [] |
| 263 | items_data = () |
| 264 | for row in table["table_info"]["rows"]: |
| 265 | item = "(" + ",".join(["%s"] * len(row)) + ")" |
| 266 | items_data += tuple(str(col) for col in row) |
| 267 | items.append(item) |
| 268 | items_str = ",".join(items) |
| 269 | final_sql.append(f'CREATE TABLE IF NOT EXISTS `{name}` ({columns})') |
| 270 | final_sql.append(( |
| 271 | f'INSERT INTO `{name}` ({column_names}) VALUES {items_str}', |
| 272 | items_data |
| 273 | )) |
| 274 | return final_sql |