feat: datasource api

Author: XiaJunjie2020

backend/apps/datasource/api/datasource.py

@@ -1,9 +1,18 @@
 from fastapi import APIRouter, HTTPException
-from ..crud.datasource import get_datasource_list
+from ..crud.datasource import get_datasource_list, check_status, create_ds
 from common.core.deps import SessionDep
+from ..models.datasource import DatasourceConf, CoreDatasource
 
 router = APIRouter(tags=["datasource"], prefix="/datasource")
 
 @router.get("/list")
 def datasource_list(session: SessionDep):
-    return get_datasource_list(session=session)
+    return get_datasource_list(session=session)
+
+@router.post("/check")
+def check(session: SessionDep, conf: DatasourceConf):
+    return check_status(session, conf)
+
+@router.post("/add", response_model=CoreDatasource)
+def check(session: SessionDep, ds: CoreDatasource):
+    return create_ds(session, ds)

backend/apps/datasource/crud/datasource.py

@@ -1,7 +1,39 @@
 from sqlmodel import Session, select
-from ..models.datasource import core_datasource
+from ..models.datasource import CoreDatasource, DatasourceConf
+import pyodbc
+import datetime
 
-def get_datasource_list(session: Session) -> core_datasource | None:
-    statement = select(core_datasource)
+def get_datasource_list(session: Session) -> CoreDatasource:
+    statement = select(CoreDatasource)
     datasource_list = session.exec(statement).fetchall()
-    return datasource_list
+    return datasource_list
+
+def check_status(session: Session, conf: DatasourceConf):
+    conn_str = (
+        "DRIVER={MySQL ODBC 9.3 Unicode Driver};" # todo driver config
+        f"SERVER={conf.host};"
+        f"DATABASE={conf.database};"
+        f"UID={conf.username};"
+        f"PWD={conf.password};"
+        f"PORT={conf.port};"
+    )
+
+    conn = None
+    try:
+        conn = pyodbc.connect(conn_str)
+        print("Connect Success")
+        return True
+    except pyodbc.Error as e:
+        print(f"Connect Fail：{e}")
+        return False
+    finally:
+        if conn is not None:
+            conn.close()
+
+def create_ds(session: Session, ds: CoreDatasource):
+    record = CoreDatasource(**ds.model_dump())
+    record.create_time = datetime.datetime.now()
+    record.status = "Success" # todo check status
+    session.add(record)
+    session.commit()
+    return record

backend/apps/datasource/models/datasource.py

@@ -1,13 +1,23 @@
 from sqlmodel import SQLModel, Field
 from sqlalchemy import Column, Text, BigInteger, DateTime, Integer, Identity
 from datetime import datetime
+from pydantic import BaseModel
 
-class core_datasource(SQLModel, table=True):
+class CoreDatasource(SQLModel, table=True):
+    __tablename__ = "core_datasource"
     id: int = Field(sa_column=Column(Integer, Identity(always=True), nullable=False, primary_key=True))
     name: str = Field(max_length=128, nullable=False)
     description: str = Field(max_length=512, nullable=True)
     type: str = Field(max_length=64)
     configuration: str = Field(sa_column=Column(Text))
     create_time: datetime = Field(sa_column=Column(DateTime(timezone=True), nullable=True))
     create_by: int = Field(sa_column=Column(BigInteger()))
-    status: str = Field(max_length=64, nullable=True)
+    status: str = Field(max_length=64, nullable=True)
+
+class DatasourceConf(BaseModel):
+    host: str
+    port: int
+    username: str
+    password: str
+    database: str
+    driver: str