ref #26 后端瘦身、引入pre-commit

.github/workflows/build.py

@@ -1,4 +1,6 @@
 import yapf_third_party
+from PyInstaller import __main__ as pyi
+
 
 params = [
     "--windowed",
@@ -10,9 +12,8 @@
     "--clean",
     "--noconfirm",
     "--name=client",
-    "main.py",
+    "server.py",
 ]
 
-from PyInstaller import __main__ as pyi
 
 pyi.run(params)

.github/workflows/build_client.yml

@@ -0,0 +1 @@
+# 构建桌面端

.pre-commit-config.yaml

@@ -0,0 +1,14 @@
+# See https://pre-commit.com for more information
+# See https://pre-commit.com/hooks.html for more hooks
+default_language_version:
+    python: python3.11
+repos:
+  - repo: https://github.com/astral-sh/ruff-pre-commit
+    # Ruff version.
+    rev: v0.4.2
+    hooks:
+      # Run the linter.
+      - id: ruff
+        args: [ --fix ]
+      # Run the formatter.
+      - id: ruff-format

dfs_generate/__init__.py

@@ -0,0 +1 @@
+__version__ = "0.2.0"

dfs_generate/conversion.py

@@ -0,0 +1,239 @@
+from string import Template
+
+from tpl import SQLMODEL_DAO, TORTOISE_DAO, RESPONSE_SCHEMA, SQLMODEL_ROUTER, MAIN
+from tools import to_pascal, tran, to_snake
+
+
+def _pydantic_field(column, imports):
+    # 列名
+    name = column["COLUMN_NAME"]
+    # 类型
+    t = column["DATA_TYPE"]
+    info = tran(t, "pydantic")
+    end = "None"
+    if desc := column["COLUMN_COMMENT"]:
+        end = f"Field(None, description='{desc}')"
+        imports.add("from pydantic import Field")
+    field = f"{name}: Optional[{info['type']}] = {end}"
+    if info["import"]:
+        imports.add(info["import"])
+    return field
+
+
+class Conversion:
+    def __init__(self, table_name, columns):
+        self.table_name = table_name
+        self.columns = columns
+
+    @property
+    def table(self):
+        return to_pascal(self.table_name)
+
+    @property
+    def router_name(self):
+        return to_snake(self.table_name)
+
+    def model(self):
+        pass
+
+    def dao(self):
+        pass
+
+    def schema(self):
+        imports = {
+            "from typing import Optional",
+            "from pydantic import BaseModel, Field",
+            "from pydantic.alias_generators import to_camel",
+        }
+        head = f"class {self.table}(BaseModel):"
+        fields = []
+        for column in self.columns:
+            field = _pydantic_field(column, imports)
+            if "    " + field not in fields:
+                fields.append("    " + field)
+        fields.append(
+            "    "
+            + 'model_config = {"alias_generator": to_camel, "populate_by_name": True}'
+        )
+        return (
+            "\n".join(imports)
+            + "\n\n"
+            + RESPONSE_SCHEMA
+            + "\n\n"
+            + head
+            + "\n"
+            + "\n".join(fields)
+            + "\n"
+        )
+
+    def router(self):
+        pass
+
+    def main(self):
+        return MAIN.format(router_name=self.router_name)
+
+    def gencode(self):
+        return {
+            "model.py": self.model(),
+            "dao.py": self.dao(),
+            "router.py": self.router(),
+            "schema.py": self.schema(),
+            "main.py": self.main(),
+        }
+
+
+def _sqlmodel_field_repr(column, imports):
+    # 列名
+    info = tran(column["DATA_TYPE"], "pydantic")
+    _type = info["type"]
+    if imported := info["import"]:
+        imports.add(imported)
+    kwargs = {"default": None}
+    if v := column["CHARACTER_MAXIMUM_LENGTH"]:
+        kwargs["max_length"] = v
+    if v := column["NUMERIC_PRECISION"] and _type == "Decimal":
+        kwargs["max_digits"] = v
+    if v := column["NUMERIC_SCALE"] and _type == "Decimal":
+        kwargs["decimal_places"] = v
+    if _type == "dict":
+        imports.add("from sqlmodel import JSON")
+        kwargs["sa_type"] = "JSON"
+
+    if column["IS_NULLABLE"] == "YES":
+        kwargs["nullable"] = True
+    else:
+        kwargs["default"] = "..."
+
+    if v := column["COLUMN_DEFAULT"]:
+        if _type == "int":
+            kwargs["default"] = v
+
+    # 主键
+    if column["COLUMN_KEY"] == "PRI":
+        kwargs["primary_key"] = True
+        kwargs["default"] = None
+
+    # 描述
+    if desc := column["COLUMN_COMMENT"]:
+        kwargs["description"] = '"' + desc + '"'
+
+    if column["EXTRA"] == "DEFAULT_GENERATED":
+        imports.add("from datetime import datetime")
+        kwargs.update({"default_factory": "datetime.utcnow"})
+
+    elif column["EXTRA"].startswith("DEFAULT_GENERATED on update"):
+        imports.add("from sqlmodel import func, DateTime, Column")
+        kwargs.update({"sa_column": "Column(DateTime(), onupdate=func.now())"})
+
+    if column["COLUMN_KEY"] == "MUL":
+        kwargs["index"] = True
+
+    if column["COLUMN_KEY"] == "UNI":
+        kwargs["unique"] = True
+
+    if "default_factory" in kwargs:
+        kwargs.pop("default")
+
+    if "sa_column" in kwargs and "nullable" in kwargs:
+        kwargs.pop("nullable")
+
+    name = column["COLUMN_NAME"]
+    if kwargs.get("default", "") is None and "func.now" not in kwargs.get(
+        "sa_column", ""
+    ):
+        imports.add("from typing import Optional")
+        field_type = f"Optional[{_type}]"
+    else:
+        field_type = info["type"]
+    head = f"{name}: {field_type} = Field("
+    tail = ",".join(f"{k}={v}" for k, v in kwargs.items()) + ")"
+    return head + tail
+
+
+class SQLModelConversion(Conversion):
+    def model(self):
+        imports = {"from sqlmodel import SQLModel, Field"}
+        head = f"class {self.table}(SQLModel, table=True):"
+        fields = []
+        for column in self.columns:
+            field = _sqlmodel_field_repr(column, imports)
+            if "    " + field not in fields:
+                fields.append("    " + field)
+        return "\n".join(imports) + "\n\n" + head + "\n" + "\n".join(fields)
+
+    def dao(self):
+        imports = {
+            "from sqlmodel import Session, SQLModel,select, func, Field",
+            "from typing import List, Optional",
+            "import model",
+            "import schema",
+        }
+        content = SQLMODEL_DAO.format(table=self.table)
+        return "\n".join(imports) + "\n\n" + content
+
+    def router(self):
+        return Template(SQLMODEL_ROUTER).safe_substitute(
+            router_name=self.router_name, table=self.table
+        )
+
+
+def _tortoise_field_repr(column):
+    name = column["COLUMN_NAME"]
+    info = tran(column["DATA_TYPE"], "tortoise-orm")
+    kwargs = {}
+    if v := column["IS_NULLABLE"] == "YES":
+        kwargs["null"] = v
+    if v := column["CHARACTER_MAXIMUM_LENGTH"]:
+        kwargs["max_length"] = v
+
+    if v := column["NUMERIC_PRECISION"] and info["type"] == "Decimal":
+        kwargs["max_digits"] = v
+
+    if v := column["NUMERIC_SCALE"] and info["type"] == "Decimal":
+        kwargs["decimal_places"] = v
+
+    if v := column["COLUMN_DEFAULT"]:
+        if info["type"] == "int":
+            kwargs["default"] = v
+        else:
+            kwargs["default"] = f"{v}"
+    if v := column["COLUMN_COMMENT"]:
+        kwargs["description"] = f'"{v}"'
+
+    if column["COLUMN_KEY"] == "MUL":
+        kwargs["index"] = True
+
+    if column["EXTRA"] == "DEFAULT_GENERATED":
+        kwargs["auto_now_add"] = True
+        if "default" in kwargs:
+            kwargs.pop("default")
+
+    if column["EXTRA"].startswith("DEFAULT_GENERATED on update"):
+        kwargs["auto_now"] = True
+        if "default" in kwargs:
+            kwargs.pop("default")
+
+    if column["COLUMN_KEY"] == "UNI":
+        kwargs["unique"] = True
+
+    if column["COLUMN_KEY"] == "PRI":
+        kwargs["pk"] = True
+
+    return f"{name} = fields.{info['type']}({', '.join(f'{k}={v}' for k, v in kwargs.items())})"
+
+
+class TortoiseConversion(Conversion):
+    def model(self):
+        imports = {"from tortoise import Model, fields"}
+        head = f"class {self.table}(Model):"
+        fields = []
+        for column in self.columns:
+            field = _tortoise_field_repr(column)
+            if "    " + field not in fields:
+                fields.append("    " + field)
+        return "\n".join(imports) + "\n\n" + head + "\n" + "\n".join(fields)
+
+    def dao(self):
+        imports = {"from typing import List, Optional", "import model", "import schema"}
+        content = TORTOISE_DAO.format(table=self.table)
+        return "\n".join(imports) + "\n\n" + content

dfs_generate/server.py

@@ -0,0 +1,101 @@
+import os
+from typing import Dict
+
+import bottle
+import isort
+from yapf.yapflib.yapf_api import FormatCode
+
+from conversion import SQLModelConversion, TortoiseConversion
+from tools import MySQLConf, MySQLHelper
+from tpl import SQLMODEL_DB
+
+app = bottle.Bottle()
+
+CACHE: Dict[str, MySQLHelper] = {}
+
+# 解决打包桌面程序static找不到的问题
+static_file_abspath = os.path.join(os.path.dirname(__file__), "../web/dist")
+
+
+@app.hook("before_request")
+def validate():
+    request_method = bottle.request.environ.get("REQUEST_METHOD")
+    access_control = bottle.request.environ.get("HTTP_ACCESS_CONTROL_REQUEST_METHOD")
+    if request_method == "OPTIONS" and access_control:
+        bottle.request.environ["REQUEST_METHOD"] = access_control
+
+
+@app.hook("after_request")
+def enable_cors():
+    bottle.response.headers["Access-Control-Allow-Origin"] = "*"
+    bottle.response.headers["Access-Control-Allow-Headers"] = "*"
+
+
+# 定义路由，提供静态文件服务
+@app.get("/static/<filepath:path>")
+def serve_static(filepath):
+    return bottle.static_file(filepath, root=static_file_abspath)
+
+
+# 定义首页路由
+@app.get("/")
+def index():
+    with open(os.path.join(static_file_abspath, "index.html"), "rb") as f:
+        html_content = f.read()
+    bottle.response.content_type = "text/html"
+    return html_content
+
+
+@app.post("/conf")
+def connect():
+    payload = bottle.request.json
+    if payload:
+        conf = MySQLConf(**payload)
+        CACHE["connect"] = MySQLHelper(conf)
+        return {"code": 20000, "msg": "ok", "data": None}
+
+
+@app.get("/tables")
+def tables():
+    if obj := CACHE.get("connect"):
+        like = bottle.request.query.get("tableName")
+        data = [
+            {"tableName": table, "key": table}
+            for table in obj.get_tables()
+            if like in table
+        ]
+        return {"code": 20000, "msg": "ok", "data": data}
+    return []
+
+
+@app.get("/codegen")
+def codegen():
+    obj = CACHE.get("connect")
+    table = bottle.request.query.get("tableName")
+    mode = bottle.request.query.get("mode")
+    results = []
+    if mode == "sqlmodel":
+        data = SQLModelConversion(table, obj.get_table_columns(table)).gencode()
+
+        for k, v in data.items():
+            _code = FormatCode(v, style_config="pep8")[0]
+            v = isort.code(_code)
+            results.append({"name": k, "code": v, "key": k})
+        results.append(
+            {
+                "name": "db.py",
+                "code": SQLMODEL_DB.format(uri=obj.conf.get_db_uri()),
+                "key": "db.py",
+            }
+        )
+    else:
+        data = TortoiseConversion(table, obj.get_table_columns(table)).gencode()
+        for k, v in data.items():
+            _code = FormatCode(v, style_config="pep8")[0]
+            v = isort.code(_code)
+            results.append({"name": k, "code": v, "key": k})
+    return {"code": 20000, "msg": "ok", "data": results}
+
+
+if __name__ == "__main__":
+    app.run(host="0.0.0.0", port=8080, debug=True, reloader=True)