数据库连接配置
1
2
3
4
5
6
7
8
9
|
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 |
创建实体类
1
2
3
4
5
6
7
8
9
10
11
12
|
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:
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
|
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语句
1
2
3
4
5
6
7
8
9
10
11
12
|
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 |
或、与、非和排序
1
2
3
4
5
6
7
8
9
10
11
12
13
|
@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 ) |
到此这篇关于详解Flask数据库的连接与使用的文章就介绍到这了,更多相关Flask数据库内容请搜索服务器之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持服务器之家!
原文链接:https://wtl4it.blog.csdn.net/article/details/129107801