datax-admin/app/core/datasource/datasource.py

from dataclasses import dataclass
from typing import Optional
from app.utils import *


@dataclass
class DataSourceBase:
    type: str
    host: Optional[str]
    port: Optional[int]
    username: str
    password: str
    database_name: str

    @property
    def jdbc_username(self):
        return encode_base64(self.username)


    @property
    def jdbc_password(self):
        return encode_base64(self.password)

    @property
    def jdbc_url(self):
        pass

    @property
    def jdbc_driver_class(self):
        pass

    @property
    def connection_str(self):
        pass

    def is_connect(self):
        pass

    def get_preview_data(self, table_name, limit=100):
        pass

    def list_tables(self):
        pass

    def get_table_schema(self, table_name):
        pass


class DataSrouceFactory:
    @staticmethod
    def create(ds_type: str, ds_config: dict):
        if ds_type == 'mysql':
            from .mysql import MysqlDS
            return MysqlDS(**ds_config, type=ds_type)
        elif ds_type == 'hive':
            from .hive import HiveDS
            return HiveDS(**ds_config, type=ds_type)
        else:
            raise Exception('不支持的数据源类型')