全部目录结构完成

.env

@@ -0,0 +1,3 @@
+ALGORITHM=HS256
+DATABASE_URL=sqlite:///test.db
+SECRET_KEY=your_secret_key

api/__init__.py

@@ -0,0 +1,5 @@
+# -*- coding: utf-8 -*-
+# @Time    : 2024/11/19 下午7:26
+# @Author  : 河瞬
+# @FileName: __init__.py.py
+# @Software: PyCharm

api/login_reg.py

@@ -0,0 +1,51 @@
+# -*- coding: utf-8 -*-
+# @Time    : 2024/11/19 下午7:27
+# @Author  : 河瞬
+# @FileName: login_reg.py
+# @Software: PyCharm
+from fastapi import HTTPException, Response, Depends, APIRouter
+from typing import Optional, Annotated
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+
+from sqlmodel import select
+
+from models import Tenant, User, Project
+from dependencies import *
+
+router = APIRouter()
+
+
+# 生成JWT token
+def create_access_token(data: dict, expires_delta: Optional[timedelta] = None, settings: SettingsDep = SettingsDep):
+    to_encode = data.copy()
+    if expires_delta:
+        expire = datetime.utcnow() + expires_delta
+    else:
+        expire = datetime.utcnow() + timedelta(minutes=15)
+    to_encode.update({"exp": expire})
+    print(settings, type(settings))
+    encoded_jwt = jwt.encode(to_encode, settings.SECRET_KEY, algorithm=settings.ALGORITHM)
+    return encoded_jwt
+
+
+# 登录路由
+@router.post("/api/s1/login")
+async def login(response: Response, user_data: dict, session: SessionDep):
+    # 查询用户
+    user = session.exec(select(User).where(User.username == user_data['username'])).first()
+
+    # 验证用户名和密码
+    if not user or user.password != user_data['password']:
+        raise HTTPException(status_code=401, detail="Login failed")
+
+    # 生成JWT token
+    token = create_access_token(data={"id": user.id, "role": user.role, "tanant_id": user.tenant.id})
+
+    # 设置cookie
+    response.set_cookie(key="session_token", value=token, httponly=True)
+
+    # 关闭数据库会话
+    session.close()
+
+    return {"message": f"Login successful"}

api/manage_project.py

@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+# @Time    : 2024/11/19 下午8:05
+# @FileName: manage_project.py
+# @Software: PyCharm
+from fastapi import HTTPException, Response, Depends, APIRouter
+from typing import Optional, Annotated
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+
+from sqlmodel import select
+
+from models import Tenant, User, Project
+from dependencies import *
+
+router = APIRouter()
+
+
+@router.get(...)
+def example():
+    return "hello"

api/manage_tanant.py

@@ -0,0 +1,20 @@
+# -*- coding: utf-8 -*-
+# @Time    : 2024/11/19 下午8:04
+# @FileName: manage_tanant.py
+# @Software: PyCharm
+from fastapi import HTTPException, Response, Depends, APIRouter
+from typing import Optional, Annotated
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+
+from sqlmodel import select
+
+from models import Tenant, User, Project
+from dependencies import *
+
+router = APIRouter()
+
+
+@router.get(...)
+def example():
+    return "hello"

api/manage_user.py

@@ -0,0 +1,21 @@
+# -*- coding: utf-8 -*-
+# @Time    : 2024/11/19 下午8:05
+# @Author  : 河瞬
+# @FileName: manage_user.py
+# @Software: PyCharm
+from fastapi import HTTPException, Response, Depends, APIRouter
+from typing import Optional, Annotated
+from datetime import datetime, timedelta
+from jose import JWTError, jwt
+
+from sqlmodel import select
+
+from models import Tenant, User, Project
+from dependencies import *
+
+router = APIRouter()
+
+
+@router.get(...)
+def example():
+    return "hello"

config.py

@@ -0,0 +1,18 @@
+# -*- coding: utf-8 -*-
+# @Time    : 2024/11/19 下午7:30
+# @Author  : 河瞬
+# @FileName: config.py
+# @Software: PyCharm
+from pydantic_settings import BaseSettings
+
+
+class Settings(BaseSettings):
+    ALGORITHM: str = "HS256"
+    DATABASE_URL: str = "sqlite:///test.db"
+    SECRET_KEY: str = "your_secret_key"
+
+    class Config:
+        env_file = ".env"
+
+if __name__ == '__main__':
+    print(Settings().ALGORITHM)

database.py

@@ -5,8 +5,9 @@
 # @Software: PyCharm
 from sqlmodel import SQLModel, create_engine
 
-sqlite_file_name = "test.db"
+from config import Settings
-sqlite_url = f"sqlite:///{sqlite_file_name}"
+
+sqlite_url = Settings().DATABASE_URL
 
 engine = create_engine(sqlite_url)
 

dependencies.py

@@ -0,0 +1,24 @@
+# -*- coding: utf-8 -*-
+# @Time    : 2024/11/19 下午7:33
+# @Author  : 河瞬
+# @FileName: dependencies.py
+# @Software: PyCharm
+from typing import Annotated
+from fastapi import Depends
+from database import engine
+from sqlmodel import Session
+from config import Settings
+
+
+def get_session():
+    with Session(engine) as session:
+        yield session
+
+
+def get_settings():
+    return Settings()
+
+
+SessionDep = Annotated[Session, Depends(get_session)]
+
+SettingsDep = get_settings()

main.py

@@ -9,6 +9,8 @@ from sqlmodel import Session, select
 
 from database import create_db_and_tables, engine
 from models import Tenant, User, Project
+from dependencies import *
+from api import login_reg, manage_project, manage_tanant, manage_user
 
 # 用于生成和验证JWT的密钥
 SECRET_KEY = "your_secret_key"
@@ -24,44 +26,8 @@ async def lifespan(app: FastAPI):
     yield
 
 
-def get_session():
-    with Session(engine) as session:
-        yield session
-
-
-# 生成JWT token
-def create_access_token(data: dict, expires_delta: Optional[timedelta] = None):
-    to_encode = data.copy()
-    if expires_delta:
-        expire = datetime.utcnow() + expires_delta
-    else:
-        expire = datetime.utcnow() + timedelta(minutes=15)
-    to_encode.update({"exp": expire})
-    encoded_jwt = jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)
-    return encoded_jwt
-
-
 app = FastAPI(lifespan=lifespan)
-SessionDep = Annotated[Session, Depends(get_session)]
+app.include_router(login_reg.router)
-
+app.include_router(manage_tanant.router)
-
+app.include_router(manage_user.router)
-# 登录路由
+app.include_router(manage_project.router)
-@app.post("/api/s1/login")
-async def login(response: Response, user_data: dict, session: SessionDep):
-    # 查询用户
-    user = session.exec(select(User).where(User.username == user_data['username'])).first()
-
-    # 验证用户名和密码
-    if not user or user.password != user_data['password']:
-        raise HTTPException(status_code=401, detail="Login failed")
-
-    # 生成JWT token
-    token = create_access_token(data={"id": user.id, "role": user.role, "tanant_id": user.tenant.id})
-
-    # 设置cookie
-    response.set_cookie(key="session_token", value=token, httponly=True)
-
-    # 关闭数据库会话
-    session.close()
-
-    return {"message": f"Login successful"}

requirements.txt

@@ -1,4 +1,6 @@
 fastapi~=0.115.5
-SQLAlchemy~=2.0.36
 python-jose~=3.3.0
 uvicorn~=0.32.0
+pydantic~=2.9.2
+pydantic-settings~=2.6.1
+sqlmodel~=0.0.22