首次提交

app.py

@@ -0,0 +1,36 @@
+from flask import Flask
+from config import Config
+from extensions import cors, db, jwt
+from blueprints.user import user_bp
+from blueprints.auth import auth_bp
+from common.exceptions import BusinessException
+from common.response import error
+
+def create_app():
+    app = Flask(__name__)
+    app.config.from_object(Config)
+
+    cors.init_app(app)
+    db.init_app(app)
+    jwt.init_app(app)
+
+    app.register_blueprint(user_bp)
+    app.register_blueprint(auth_bp)
+
+    @app.errorhandler(BusinessException)
+    def handle_business_exception(e):
+        return error(e.code, e.message)
+
+    @app.errorhandler(Exception)
+    def handle_exception(e):
+        return error(50000, "Internal Server Error")
+
+    with app.app_context():
+        db.create_all()
+
+    return app
+
+app = create_app()
+
+if __name__ == "__main__":
+    app.run(port=5000)

blueprints/auth.py

@@ -0,0 +1,14 @@
+from flask import Blueprint, request
+from common.response import success
+from services.auth_service import AuthService
+
+auth_bp = Blueprint("auth", __name__, url_prefix="/api/auth")
+
+@auth_bp.post("/login")
+def login():
+    body = request.get_json()
+    username = body.get("username")
+    password = body.get("password")
+
+    data = AuthService.login(username, password)
+    return success(data)

blueprints/user.py

@@ -0,0 +1,18 @@
+from flask import Blueprint
+from flask_jwt_extended import jwt_required, get_jwt_identity
+from common.response import success
+from services.user_service import UserService
+
+user_bp = Blueprint("user", __name__, url_prefix="/api/users")
+
+@user_bp.get("")
+@jwt_required()
+def list_users():
+    data = UserService.list_users()
+    return success(data)
+
+@user_bp.get("/me")
+@jwt_required()
+def me():
+    user_id = get_jwt_identity()
+    return success({"user_id": user_id})

common/exceptions.py

@@ -0,0 +1,4 @@
+class BusinessException(Exception):
+    def __init__(self, code, message):
+        self.code = code
+        self.message = message

common/response.py

@@ -0,0 +1,15 @@
+from flask import jsonify
+
+def success(data=None, message="success"):
+    return jsonify({
+        "code": 0,
+        "message": message,
+        "data": data
+    })
+
+def error(code=50000, message="error"):
+    return jsonify({
+        "code": code,
+        "message": message,
+        "data": None
+    })

config.py

@@ -0,0 +1,15 @@
+class Config:
+    DEBUG = True
+
+    DATABASE_USER = "postgres"
+    DATABASE_PASSWORD = "root"
+    DATABASE_HOST = "localhost"
+    DATABASE_PORT = "5432"
+    DATABASE_NAME = "postgres"
+    SQLALCHEMY_DATABASE_URI = (
+        f"postgresql+psycopg2://{DATABASE_USER}:{DATABASE_PASSWORD}@{DATABASE_HOST}:{DATABASE_PORT}/{DATABASE_NAME}"
+    )
+
+    SQLALCHEMY_TRACK_MODIFICATIONS = False
+
+    JWT_SECRET_KEY = "114514"

extensions.py

@@ -0,0 +1,7 @@
+from flask_cors import CORS
+from flask_sqlalchemy import SQLAlchemy
+from flask_jwt_extended import JWTManager
+
+cors = CORS()
+db = SQLAlchemy()
+jwt = JWTManager()

models/user.py

@@ -0,0 +1,28 @@
+from extensions import db  # 从扩展模块导入初始化好的 SQLAlchemy 数据库实例
+
+
+# 定义 User 类，继承自 db.Model，使其成为一个数据库模型类（对应数据库的一张表）
+class User(db.Model):
+    # 【表配置】显式指定在数据库中生成的表名为 "users"
+    __tablename__ = "users"
+
+    # 【字段定义】
+    # id：字段名；db.Integer：整数类型；primary_key=True：设为主键（通常会自动自增）
+    id = db.Column(db.Integer, primary_key=True)
+
+    # username：用户名；db.String(64)：最大长度64的字符串
+    # unique=True：值必须唯一（不可重复）；nullable=False：不允许为空（必填项）
+    username = db.Column(db.String(64), unique=True, nullable=False)
+
+    # password：密码；db.String(128)：最大长度128（通常存加密后的哈希串）
+    # nullable=False：不允许为空
+    password = db.Column(db.String(128), nullable=False)
+
+    # 【实例方法：序列化】
+    # 将数据库对象转换为 Python 字典，方便在 Web 开发中返回 JSON 数据给前端
+    def to_dict(self):
+        return {
+            "id": self.id,
+            "username": self.username
+            # 注意：出于安全考虑，通常不会在 to_dict 中返回 password 字段
+        }

repositories/user_repo.py

@@ -0,0 +1,19 @@
+from models.user import User
+from extensions import db
+
+class UserRepository:
+
+    @staticmethod
+    def find_by_username(username: str):
+        return User.query.filter_by(username=username).first()
+
+    @staticmethod
+    def find_by_id(user_id: int):
+        return User.query.get(user_id)
+
+    @staticmethod
+    def create(username: str, password: str):
+        user = User(username=username, password=password)
+        db.session.add(user)
+        db.session.commit()
+        return user

requirements.txt

@@ -0,0 +1,4 @@
+Flask==3.0.0
+Flask-CORS==4.0.0
+Flask-SQLAlchemy==3.1.1
+Flask-JWT-Extended==4.6.0

services/auth_service.py

@@ -0,0 +1,22 @@
+from repositories.user_repo import UserRepository
+from common.exceptions import BusinessException
+from flask_jwt_extended import create_access_token
+
+class AuthService:
+
+    @staticmethod
+    def login(username: str, password: str):
+        user = UserRepository.find_by_username(username)
+
+        if not user:
+            raise BusinessException(40101, "User not found")
+
+        if user.password != password:
+            raise BusinessException(40102, "Invalid credentials")
+
+        token = create_access_token(identity=user.id)
+
+        return {
+            "token": token,
+            "user": user.to_dict()
+        }

services/user_service.py

@@ -0,0 +1,8 @@
+from repositories.user_repo import UserRepository
+
+class UserService:
+
+    @staticmethod
+    def list_users():
+        users = UserRepository.find_all()
+        return [u.to_dict() for u in users]