feat:mysql adaptation for meta db (#50)

* phase1:uuid stringdefault timestamp date

* phase2:json adapt

* phase3:BinaryData,ARRAY,Text,

* fix sa.String()

* fix defalut valud

* models adapt

* models adapt supplement

* fix pg unique dialect

* supplement

* migration fix

* migration fix

* date sql adapt

* model adapt fix

* config adapt

* migration fix

* fix

* fix anotation in .env.example

* config feat

* config fix

* .env.example adapt fix
This commit is contained in:
longbingljw
2025-11-07 15:35:31 +08:00
committed by GitHub
parent 829796514a
commit fcc71124b1
125 changed files with 6096 additions and 2427 deletions

View File

@ -2,8 +2,9 @@ import enum
import uuid
from typing import Any, Generic, TypeVar
from sqlalchemy import CHAR, VARCHAR, TypeDecorator
from sqlalchemy.dialects.postgresql import UUID
from sqlalchemy import CHAR, TEXT, VARCHAR, LargeBinary, TypeDecorator
from sqlalchemy.dialects.mysql import LONGBLOB, LONGTEXT
from sqlalchemy.dialects.postgresql import BYTEA, UUID
from sqlalchemy.engine.interfaces import Dialect
from sqlalchemy.sql.type_api import TypeEngine
@ -34,6 +35,52 @@ class StringUUID(TypeDecorator[uuid.UUID | str | None]):
return str(value)
class LongText(TypeDecorator[str | None]):
impl = TEXT
cache_ok = True
def process_bind_param(self, value: str | None, dialect: Dialect) -> str | None:
if value is None:
return value
return value
def load_dialect_impl(self, dialect: Dialect) -> TypeEngine[Any]:
if dialect.name == "postgresql":
return dialect.type_descriptor(TEXT())
elif dialect.name == "mysql":
return dialect.type_descriptor(LONGTEXT())
else:
return dialect.type_descriptor(TEXT())
def process_result_value(self, value: str | None, dialect: Dialect) -> str | None:
if value is None:
return value
return value
class BinaryData(TypeDecorator[bytes | None]):
impl = LargeBinary
cache_ok = True
def process_bind_param(self, value: bytes | None, dialect: Dialect) -> bytes | None:
if value is None:
return value
return value
def load_dialect_impl(self, dialect: Dialect) -> TypeEngine[Any]:
if dialect.name == "postgresql":
return dialect.type_descriptor(BYTEA())
elif dialect.name == "mysql":
return dialect.type_descriptor(LONGBLOB())
else:
return dialect.type_descriptor(LargeBinary())
def process_result_value(self, value: bytes | None, dialect: Dialect) -> bytes | None:
if value is None:
return value
return value
_E = TypeVar("_E", bound=enum.StrEnum)