FastAPI（六十四）实战开发《在线课程学习系统》设计数据库以及连接配置

前言

        针对前面的实战开发“在线课程学习系统”梳理系统需要的接口，我们首先对数据表进行设计。

正文

from sqlalchemy import Column, Integer, String, ForeignKey, Boolean, Text, DateTime
from sqlalchemy import MetaData
from sqlalchemy.inspection import inspect
from datetime import datetime

from sqlalchemy.orm import DeclarativeMeta

from middlewares.mysql.database import Base, engine


class Role(Base):
    """角色"""
    __tablename__ = "roles"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(length=8), unique=True, index=True)  # 角色名称


class User(Base):
    """用户基础表"""
    __tablename__ = "users"

    id = Column(Integer, primary_key=True, index=True)
    username = Column(String(length=32), unique=True, index=True)  # 用户名
    password = Column(String(length=252))  # 密码
    status = Column(Integer, default=0)  # 删除，0正常
    job_num = Column(Integer, nullable=True)  # 工号
    student_num = Column(Integer, nullable=True)  # 学号
    age = Column(Integer)  # 年龄
    sex = Column(String(length=8), default="男")  # 性别
    role = Column(Integer, ForeignKey('roles.id'))  # 角色
    add_time = Column(DateTime, default=datetime.now())


class Course(Base):
    """课程"""
    __tablename__ = "courses"

    id = Column(Integer, primary_key=True, index=True)
    name = Column(String(length=252), unique=True, index=True)  # 课程名称
    icon = Column(String(length=252), nullable=True)  # icon
    desc = Column(String(length=252), nullable=True)  # 描述
    status = Column(Boolean, default=False)  # 状态
    onsale = Column(Boolean, default=False)  # 是否上架
    catalog = Column(Text, nullable=True)  # 目录
    owner = Column(Integer, ForeignKey('users.id'))  # 拥有者
    like_num = Column(Integer, default=0)  # 点赞


class StudentCourse(Base):
    """学生课程"""
    __tablename__ = "student_courses"

    id = Column(Integer, primary_key=True, index=True)
    student = Column(Integer, ForeignKey('users.id'))  # 学生
    course = Column(Integer, ForeignKey('courses.id'))  # 课程
    add_time = Column(DateTime, default=datetime.now())
    update_time = Column(DateTime, default=datetime.now())
    status = Column(Integer, default=0)  # 1.删除，0.正常


class CourseComment(Base):
    """课程评论"""
    __tablename__ = "course_comments"

    id = Column(Integer, primary_key=True, index=True)
    course = Column(Integer, ForeignKey('courses.id'))  # 课程
    user = Column(Integer, ForeignKey('users.id'))  # 评论人
    pid = Column(Integer)  # 回复
    add_time = Column(DateTime, default=datetime.now())
    top = Column(Boolean, default=False)  # 是否置顶
    context = Column(Text)
    status = Column(Boolean, default=0)  # 1删除0正常


class Message(Base):
    __tablename__ = "messages"

    id = Column(Integer, primary_key=True, index=True)
    send_user = Column(Integer, ForeignKey('users.id'))  # 发送者
    accept_user = Column(Integer, ForeignKey('users.id'))  # 接收者
    read = Column(Boolean, default=False)  # 是否已读，接收者是否已读
    send_time = Column(String(length=252))  # 发送时间
    pid = Column(Integer)  # 回复者
    add_time = Column(DateTime, default=datetime.now())  # 添加时间
    context = Column(Text)
    status = Column(Integer, default=0)  # 1删除0正常


def create_tables():
    """表不存在则创建"""
    global_dict = globals()
    classes = [v
               for k, v in global_dict.items()
               if callable(v) and isinstance(v, DeclarativeMeta) and hasattr(v, "__tablename__")]
    metadata = MetaData()
    for _cls in classes:
        if not inspect(engine).has_table(_cls.__tablename__, schema=None, metadata=metadata):
           _cls.__table__.create(bind=engine)


if __name__ == '__main__':
    create_tables()

 设计完数据库后，我们需要对数据连接进行配置，我们使用的数据库是mysql，需要安装mysql的依赖，然后我们需要配置

from sqlalchemy import create_engine
from sqlalchemy.orm import declarative_base, sessionmaker
from settings.config import MYSQL_CONFIG

conn = "mysql+pymysql://{username}:{password}@{host}:{port}/{database}?charset=utf8".format(
    username=MYSQL_CONFIG["username"], password=MYSQL_CONFIG["password"], host=MYSQL_CONFIG["host"],
    port=MYSQL_CONFIG["port"], database=MYSQL_CONFIG["database"])
engine = create_engine(conn)

# 该类的每个实例都是一个数据库会话，该类本身还不是数据库会话，但是一旦我们创建了SessionLocal的实例，这个实例将是实际的数据库会话
SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)

# 创建数据库基类
Base = declarative_base()

其中MYSQL_CONFIG如下

import os

from starlette.config import Config

PROJECT_DIR_PATH = os.path.dirname(os.path.dirname(__file__))
config = Config(os.path.join(PROJECT_DIR_PATH, '.env'))

ENV = config.get("ENV", str, "DEV")

mysql_database = config.get("MYSQL_DB_DEV", str, "learn_onsite_system_dev")
if ENV == "TEST":
    mysql_database = config.get("MYSQL_DB_TEST", str, "learn_onsite_system_test")
elif ENV == "PRO":
    mysql_database = config.get("MYSQL_DB_PRO", str, "learn_onsite_system_pro")
else:
    pass

# mysql
MYSQL_CONFIG = {
    "host": config.get("MYSQL_HOST", str, "10.30.10.36"),
    "port": config.get("MYSQL_PORT", int, 3306),
    "username": config.get("MYSQL_USERNAME", str, "root"),
    "password": config.get("MYSQL_PASSWORD", str, "123456"),
    "database": mysql_database
}

可以看到，我们针对不同的环境设计了不同的数据库，只需要根据不同的环境修改环境变量中的环境变量即可自动切换数据库

如果不判断数据表是否存在就创建数据表的话，也可以使用下面命令：

Base.metadata.create_all(bind=engine)

数据库相关的深入的讲解可以看之前分享的，

FastAPI 学习之路（三十二）创建数据库

FastAPI 学习之路（三十三）操作数据库

FastAPI 学习之路（三十四）数据库多表操作