add InspectDb and show_create_tables

aerich/cli.py

@@ -4,6 +4,7 @@ import sys
 from configparser import ConfigParser
 from functools import wraps
 from pathlib import Path
+from typing import List
 
 import click
 from click import Context, UsageError
@@ -12,6 +13,7 @@ from tortoise.exceptions import OperationalError
 from tortoise.transactions import in_transaction
 from tortoise.utils import get_schema_sql
 
+from aerich.inspectdb import InspectDb
 from aerich.migrate import Migrate
 from aerich.utils import (
     get_app_connection,
@@ -288,6 +290,25 @@ async def init_db(ctx: Context, safe):
     click.secho(f'Success generate schema for app "{app}"', fg=Color.green)
 
 
+@cli.command(help="Introspects the database tables to standard output as TortoiseORM model.")
+@click.option(
+    "-t",
+    "--table",
+    help="Which tables to inspect.",
+    multiple=True,
+    required=False,
+)
+@click.pass_context
+@coro
+async def inspectdb(ctx: Context, table: List[str]):
+    config = ctx.obj["config"]
+    app = ctx.obj["app"]
+    connection = get_app_connection(config, app)
+
+    inspect = InspectDb(connection, table)
+    await inspect.inspect()
+
+
 def main():
     sys.path.insert(0, ".")
     cli()

aerich/inspectdb.py

@@ -0,0 +1,32 @@
+from typing import List, Optional
+
+from ddlparse import DdlParse
+from tortoise import BaseDBAsyncClient
+from tortoise.backends.mysql.client import MySQLSchemaGenerator
+
+
+class InspectDb:
+    def __init__(self, conn: BaseDBAsyncClient, tables: Optional[List[str]] = None):
+        self.conn = conn
+        self.tables = tables
+        self.DIALECT = conn.schema_generator.DIALECT
+
+    async def show_create_tables(self):
+        if self.DIALECT == MySQLSchemaGenerator.DIALECT:
+            if not self.tables:
+                sql_tables = f"SELECT table_name FROM information_schema.tables WHERE table_schema = '{self.conn.database}';"
+                ret = await self.conn.execute_query(sql_tables)
+                self.tables = map(lambda x: x[0], ret)
+            for table in self.tables:
+                sql_show_create_table = f"SHOW CREATE TABLE {table}"
+                ret = await self.conn.execute_query(sql_show_create_table)
+                yield ret[1][0]["Create Table"]
+        else:
+            raise NotImplementedError("Currently only support MySQL")
+
+    async def inspect(self):
+        ddl_list = self.show_create_tables()
+        async for ddl in ddl_list:
+            parser = DdlParse(ddl, DdlParse.DATABASE.mysql)
+            table = parser.parse()
+            print(table)