526互联

python SQLAlchemy 和 migrate的使用 以及 增删改

发布时间 2023-10-15 11:37:48作者: 一杯咖啡钱

 

安装对应的库

pip install SQLAlchemy

pip install pymysql

pip install Flask-Migrate

 

连接数据库代码，以及orm的使用

初始化： flask db init

数据迁移：lask db migrate

将数据映射到数据库中：flask db upgrade

from flask import Flask
from flask_sqlalchemy import SQLAlchemy
from flask_migrate import Migrate

app = Flask(__name__)


HOSTNAME = "localhost"
PORT = 3306
USERNAME = "root"
PASSWORD = "root"
DATABASE = "test_test"

app.config['SQLALCHEMY_DATABASE_URI'] = f"mysql+pymysql://{USERNAME}:{PASSWORD}@{HOSTNAME}:{PORT}/{DATABASE}?charset=utf8"

# SQLAlchemy 默认读取app.config 数据库连接信息
db = SQLAlchemy(app)

# 执行数据表函数
# with app.app_context():
# # 创建数据表只能执行一次，无法增加和删除数据
#     db.create_all()

# 映射数据库表
migrate = Migrate(app, db)

# 创建数据表
class User(db.Model):
    __tablename__ = "user"
    __table_args__ = {'mysql_engine': 'InnoDB'}
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    # varchar
    username = db.Column(db.String(100), nullable=False)
    password = db.Column(db.String(100), nullable=False)
    email = db.Column(db.String(100))

class Article(db.Model):
    __tablename__ = "article"
    __table_args__ = {'mysql_engine': 'InnoDB'}
    id = db.Column(db.Integer, primary_key=True, autoincrement=True)
    title = db.Column(db.String(200), nullable=False)
    content = db.Column(db.Text, nullable=False)
    #     添加作者外键
    author_id = db.Column(db.Integer, db.ForeignKey("user.id"))

    # backref 给User模型添加一个articles属性，用于获取文章列表. 可以认为backref="articles" 为左连接
    author = db.relationship("User", backref="articles")
    # author = db.relationship("User")
    # 等于artile.author = User.query.get(artile.author_id)

@app.route('/')
def hello_world():
    # 检测数据库连接是否成功
    try:
        connection = db.engine.connect()
        print("数据库连接成功！", connection)
    except Exception as e:
        print("数据库连接失败：", e)
    db.create_all()

    return 'Hello'



if __name__ == '__main__':
    app.run()

完成！！！

 

对数据表进行增删改查

@app.route('/user/add')
def user_add():
    # 创建数据对象
    user1 = User(username="张三", password='123456')
    # 将数据对象保存到session里面
    db.session.add(user1)
    # 提交数据
    db.session.commit()
    return "用户创建成功"


@app.route('/user/query')
def user_query():
    # 1、get查找，主键查找
    # my_user = User.query.get(1)
    # print(f"{my_user.id}")
    # 2、filter_by 查找
    # Query对象
    users = User.query.filter_by(username="张三")
    for users in users:
        print(users.username)
    return "数据查找成功"


@app.route('/user/update')
def update_user():
    # 1、get查找，主键查找
    # my_user = User.query.get(1)
    # print(f"{my_user.id}")
    # 2、filter_by 查找
    # Query对象
    user = User.query.filter_by(username="张三").first()
    user.password = "222222"
    db.session.commit()
    return "数据修改成功"


@app.route('/user/delete')
def delete_user():
    user = User.query.get(2)
    db.session.delete(user)
    db.session.commit()
    return '数据删除成功'


@app.route('/article/add')
def article_add():
    article1 = Article(title="FLASK", content="flaskXXXXX")
    article1.author = User.query.get(1)

    article2 = Article(title="Django", content="DjangoXXXXX")
    article2.author = User.query.get(1)

    db.session.add_all([article1, article2])
    db.session.commit()
    return "文章添加成功"


@app.route('/article/query')
def query_article():
    user = User.query.get(1)
    print(user.articles)
    for article in user.articles:
        print(article.title)
    return "文章查找"