详解Flask数据库的连接与使用

目录

• 数据库连接配置
• 创建实体类
• 配置打印SQL语句
• 或、与、非和排序

数据库连接配置

HOST = \"XXXXXXXXXXXXX\"
PORT = 3310
USERNAME = \"root\"
PASSWORD = \"@XXXXXXXXXXX\"
DATABASE = \"mydb\"
 
SQLALCHEMY_DATABASE_URI = f\"mysql+pymysql://{USERNAME}:{quote(PASSWORD)}@{HOST}:{PORT}/{DATABASE}?charset=utf8mb4\"
SQLALCHEMY_TRACK_MODIFICATIONS = False
SQLALCHEMY_ECHO = True

创建实体类

from exts.DBServer import db
from sqlalchemy import Column, Integer, String, Date, DateTime
 
 
class Article(db.Model):
    __tablename__ = \"article\"
    id = Column(Integer, primay_key=True, autoincrement=True)
    title = Column(String(100), nullable=True)
    pub_time = Column(DateTime, nullable=True)
    author = Column(String(100), nullable=True)
    content = Column(String(10000), nullable=True)
    origin = Column(String(1000), nullable=True)

controller:

import json
 
from flask.blueprints import Blueprint
from exts.DBServer import db
from ..model.Article import Article
from flask_sqlalchemy.query import Query
from flask_restful import marshal
from flask_restful import fields
 
article_bp = Blueprint(\"article\", __name__, url_prefix=\"/article\")
 
article_fields = {
    \"id\": fields.Integer,
    \"title\": fields.String,
    \"pub_time\": fields.DateTime,
    \"author\": fields.String,
    \"content\": fields.String,
    \"origin\": fields.String
}
 
 
@article_bp.route(\"/queryAll\")
def queryAll():
    query: Query = Article.query
    articles = query.all()
    article = query.get(1)
    article2 = query.filter_by(author=\"XXX\")
    return json.dumps(marshal(articles, fields=article_fields),ensure_ascii=False)

配置打印SQL语句

from exts.DBServer import db
from sqlalchemy import Column, Integer, String
 
 
class User(db.Model):
    __tablename__ = \"user\"
    id = Column(Integer, primary_key=True, autoincrement=True)
    username = Column(String(100), nullable=True)
    password = Column(String(100), nullable=True)
 
    def __repr__(self):
        return \"User %r\" % self.body

或、与、非和排序

@user_bp.route(\"/query\")
def query_match():
    query: Query = User.query
    result = query.filter(or_(User.username.contains(\"祥\"), User.id == 1))
 
    return json.dumps(marshal(result.all(), fields=user_fields), ensure_ascii=False)
 
 
@user_bp.route(\"/in\")
def in_sql():
    query: Query = User.query
    result = query.order_by(-User.password, -User.id)
    return json.dumps(marshal(result.all(), fields=user_fields), ensure_ascii=False)