Flask无ORM开发CRUD应用：数据库交互位置与代码分层设计咨询

Hey there! Great question about structuring your Flask CRUD app without an ORM—this is exactly the kind of separation of concerns that makes apps maintainable long-term. Let’s break this down step by step, with concrete changes to your existing code.

Core Principle: Don’t Mix Concerns #

First rule of thumb: never put database operations directly in your routes (views.py). Also, keep your User entity as a pure data container, and split business logic and database access into dedicated layers. Here’s the breakdown of each layer’s role:

• Route/View Layer: Handles HTTP requests, validates input, returns responses. No database logic here—only call into your service layer.
• Service Layer: Implements business rules (e.g., password hashing, active state checks, activity tracking). Delegates database work to a repository.
• Repository Layer: Dedicated to all database interactions. Writes SQL, manages connections, and executes CRUD operations.
• Entity Layer (User class): Only holds user attributes and basic getters. No business logic or database calls—just a data structure.

Step 1: Add a Repository Layer for Database Work #

Create a new file blueprints/user/repository.py to encapsulate all user-related database operations:

from prepsmarter.extensions import conn
import pymysql

class UserRepository:
    def create_user(self, user_data):
        """Insert a new user into the database"""
        sql = """
            INSERT INTO users (email, password, registration_date, active, sign_in_count, current_sign_in_on, last_sign_in_on)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
        """
        try:
            cursor = conn.cursor()
            cursor.execute(sql, (
                user_data['email'],
                user_data['password'],
                user_data['registration_date'],
                user_data['active'],
                user_data['sign_in_count'],
                user_data['current_sign_in_on'],
                user_data['last_sign_in_on']
            ))
            conn.commit()
            return cursor.lastrowid  # Return new user ID for confirmation
        except pymysql.MySQLError as e:
            conn.rollback()
            raise e  # Propagate error to service layer for handling
        finally:
            cursor.close()

    # Add other CRUD methods here: get_user_by_email, update_user, deactivate_user, etc.

Step 2: Refactor the Service Layer #

Update blueprints/user/models.py to move business logic to UserService and keep User as a pure data entity:

import datetime
from werkzeug.security import generate_password_hash
from .repository import UserRepository

class UserService():
    def __init__(self):
        self.user_repo = UserRepository()

    def register_user(self, email, raw_password):
        """Handle registration business logic and delegate to repository"""
        # Business rule: Hash passwords before storing
        hashed_password = generate_password_hash(raw_password)
        
        # Set default values for new users
        now = datetime.datetime.now()
        user_data = {
            'email': email,
            'password': hashed_password,
            'registration_date': now,
            'active': True,
            'sign_in_count': 0,
            'current_sign_in_on': now,
            'last_sign_in_on': None
        }

        # Delegate database work to the repository
        return self.user_repo.create_user(user_data)

    def deactivate_user(self, email):
        """Deactivate a user with business checks"""
        user = self.user_repo.get_user_by_email(email)
        if not user:
            raise ValueError("User not found")
        if not user['active']:
            print(f"User {email} is already inactive")
            return
        
        self.user_repo.update_user(email, {'active': False})

    # Add other business methods: reactivate_user, update_activity_tracking, etc.

class User():
    """Pure data entity—only holds user attributes"""
    def __init__(self, email, password, registration_date, active, sign_in_count, current_sign_in_on, last_sign_in_on ):
        self.email = email
        self.password = password
        self.registration_date = registration_date
        self.active = active
        self.sign_in_count = sign_in_count
        self.current_sign_in_on = current_sign_in_on
        self.last_sign_in_on = last_sign_in_on

    def is_active(self):
        return self.active

    def __str__(self):
        user_attributes = vars(self)
        return ', '.join(f"{k}: {v}" for k, v in user_attributes.items())

Step 3: Clean Up the Route Layer #

Update blueprints/user/views.py to only handle HTTP logic and call the service layer:

from flask import Blueprint, render_template, request, jsonify
from .models import UserService

user = Blueprint('user', __name__, template_folder='templates')
user_service = UserService()

@user.route('/register')
def register_page():
    return render_template('register.html')

@user.route('/new-user', methods=['POST'])
def register_user():
    # Extract and validate input
    email = request.form.get('email')
    password = request.form.get('password')
    
    if not email or not password:
        return jsonify({"error": "Email and password are required"}), 400
    
    try:
        # Delegate registration to the service layer
        user_id = user_service.register_user(email, password)
        return jsonify({"message": "User registered successfully", "user_id": user_id}), 201
    except Exception as e:
        return jsonify({"error": str(e)}), 500

Step 4: Improve Database Connection Management #

Your current Database class can be enhanced with a context manager to auto-handle connections/prevent leaks:

import pymysql
from contextlib import contextmanager

class Database:
    _instance = None

    def __new__(cls, host, user, password, db):
        if not cls._instance:
            cls._instance = super().__new__(cls)
            cls._instance.host = host
            cls._instance.user = user
            cls._instance.password = password
            cls._instance.db = db
        return cls._instance

    @contextmanager
    def get_connection(self):
        conn = None
        try:
            conn = pymysql.connect(
                host=self.host,
                user=self.user,
                passwd=self.password,
                db=self.db
            )
            yield conn
        except Exception as e:
            if conn:
                conn.rollback()
            raise e
        finally:
            if conn:
                conn.close()

Update your repository to use this:

from prepsmarter.extensions import db  # Now db is an instance of Database

class UserRepository:
    def create_user(self, user_data):
        sql = """
            INSERT INTO users (email, password, registration_date, active, sign_in_count, current_sign_in_on, last_sign_in_on)
            VALUES (%s, %s, %s, %s, %s, %s, %s)
        """
        with db.get_connection() as conn:
            cursor = conn.cursor()
            cursor.execute(sql, tuple(user_data.values()))
            conn.commit()
            return cursor.lastrowid

Why This Structure Works #

• Testability: Mock the repository when testing the service layer—no need for a real database.
• Maintainability: If you switch databases later, only update the repository layer.
• Readability: Anyone can immediately find where business logic vs. database logic lives.

内容的提问来源于stack exchange，提问作者Pierre-Alexandre