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

振进

数据库连接配置

1. HOST = "XXXXXXXXXXXXX"
   
2. PORT = 3310
   
3. USERNAME = "root"
   
4. PASSWORD = "@XXXXXXXXXXX"
   
5. DATABASE = "mydb"
   
6. 
   
7. SQLALCHEMY_DATABASE_URI = f"mysql+pymysql://{USERNAME}:{quote(PASSWORD)}@{HOST}:{PORT}/{DATABASE}?charset=utf8mb4"
   
8. SQLALCHEMY_TRACK_MODIFICATIONS = False
   
9. SQLALCHEMY_ECHO = True

复制代码

创建实体类

1. from exts.DBServer import db
   
2. from sqlalchemy import Column, Integer, String, Date, DateTime
   
3. 
   
4. 
   
5. class Article(db.Model):
   
6.     __tablename__ = "article"
   
7.     id = Column(Integer, primay_key=True, autoincrement=True)
   
8.     title = Column(String(100), nullable=True)
   
9.     pub_time = Column(DateTime, nullable=True)
   
10.     author = Column(String(100), nullable=True)
    
11.     content = Column(String(10000), nullable=True)
    
12.     origin = Column(String(1000), nullable=True)

复制代码

controller:

1. import json
   
2. 
   
3. from flask.blueprints import Blueprint
   
4. from exts.DBServer import db
   
5. from ..model.Article import Article
   
6. from flask_sqlalchemy.query import Query
   
7. from flask_restful import marshal
   
8. from flask_restful import fields
   
9. 
   
10. article_bp = Blueprint("article", __name__, url_prefix="/article")
    
11. 
    
12. article_fields = {
    
13.     "id": fields.Integer,
    
14.     "title": fields.String,
    
15.     "pub_time": fields.DateTime,
    
16.     "author": fields.String,
    
17.     "content": fields.String,
    
18.     "origin": fields.String
    
19. }
    
20. 
    
21. 
    
22. @article_bp.route("/queryAll")
    
23. def queryAll():
    
24.     query: Query = Article.query
    
25.     articles = query.all()
    
26.     article = query.get(1)
    
27.     article2 = query.filter_by(author="XXX")
    
28.     return json.dumps(marshal(articles, fields=article_fields),ensure_ascii=False)

复制代码

配置打印SQL语句

1. from exts.DBServer import db
   
2. from sqlalchemy import Column, Integer, String
   
3. 
   
4. 
   
5. class User(db.Model):
   
6.     __tablename__ = "user"
   
7.     id = Column(Integer, primary_key=True, autoincrement=True)
   
8.     username = Column(String(100), nullable=True)
   
9.     password = Column(String(100), nullable=True)
   
10. 
    
11.     def __repr__(self):
    
12.         return "User %r" % self.body

复制代码

或、与、非和排序

1. @user_bp.route("/query")
   
2. def query_match():
   
3.     query: Query = User.query
   
4.     result = query.filter(or_(User.username.contains("祥"), User.id == 1))
   
5. 
   
6.     return json.dumps(marshal(result.all(), fields=user_fields), ensure_ascii=False)
   
7. 
   
8. 
   
9. @user_bp.route("/in")
   
10. def in_sql():
    
11.     query: Query = User.query
    
12.     result = query.order_by(-User.password, -User.id)
    
13.     return json.dumps(marshal(result.all(), fields=user_fields), ensure_ascii=False)

复制代码

到此这篇关于详解Flask数据库的连接与使用的文章就介绍到这了,更多相关Flask数据库内容请搜索脚本之家以前的文章或继续浏览下面的相关文章希望大家以后多多支持脚本之家！

来源:https://www.jb51.net/article/276387.htm
免责声明：由于采集信息均来自互联网，如果侵犯了您的权益，请联系我们【E-Mail:cb@itdo.tech】 我们会及时删除侵权内容，谢谢合作！