59 lines
2.1 KiB
Python
59 lines
2.1 KiB
Python
from fastapi import APIRouter, Depends, HTTPException, status
|
|
from sqlalchemy.orm import Session
|
|
from typing import List
|
|
from ..core.database import get_db
|
|
from ..schemas.database import (
|
|
SQLExecuteRequest, SQLExecuteResponse, TableDataRequest,
|
|
TableDataResponse, CreateTableRequest, ImportDataRequest
|
|
)
|
|
from ..services.database import DatabaseService
|
|
|
|
router = APIRouter(prefix="/database", tags=["数据库管理"])
|
|
|
|
@router.post("/execute", response_model=SQLExecuteResponse)
|
|
def execute_sql(request: SQLExecuteRequest, db: Session = Depends(get_db)):
|
|
"""执行SQL语句"""
|
|
result = DatabaseService.execute_sql(db, request.sql)
|
|
return SQLExecuteResponse(**result)
|
|
|
|
@router.post("/table-data", response_model=TableDataResponse)
|
|
def get_table_data(request: TableDataRequest, db: Session = Depends(get_db)):
|
|
"""获取表数据"""
|
|
result = DatabaseService.get_table_data(
|
|
db,
|
|
request.table_name,
|
|
request.limit or 100,
|
|
request.offset or 0
|
|
)
|
|
return TableDataResponse(**result)
|
|
|
|
@router.post("/create-table", response_model=SQLExecuteResponse)
|
|
def create_table(request: CreateTableRequest, db: Session = Depends(get_db)):
|
|
"""创建表"""
|
|
result = DatabaseService.create_table(
|
|
db,
|
|
request.table_name,
|
|
request.columns,
|
|
request.primary_key
|
|
)
|
|
return SQLExecuteResponse(**result)
|
|
|
|
@router.post("/drop-table")
|
|
def drop_table(request: dict, db: Session = Depends(get_db)):
|
|
"""删除表"""
|
|
table_name = request.get("table_name")
|
|
if not table_name:
|
|
raise HTTPException(status_code=400, detail="table_name is required")
|
|
result = DatabaseService.drop_table(db, table_name)
|
|
return SQLExecuteResponse(**result)
|
|
|
|
@router.post("/import-data", response_model=SQLExecuteResponse)
|
|
def import_data(request: ImportDataRequest, db: Session = Depends(get_db)):
|
|
"""导入数据"""
|
|
result = DatabaseService.import_data(db, request.table_name, request.data)
|
|
return SQLExecuteResponse(**result)
|
|
|
|
@router.post("/tables", response_model=List[str])
|
|
def get_table_list():
|
|
"""获取所有表名"""
|
|
return DatabaseService.get_table_list() |