前言
之前用docker编写了一个flask的简单demo,但是里面的mysql用的本地的,当时是自己偷懒用了本地的,现在空余时间重新处理了mysql容器重新更新一下
一、docker-compose.yml示例
version: '3'
services:
flask:
build:
context: ./app
container_name: flask
ports:
- "5001:5000" # Flask's internal port mapped to host port 5001
volumes:
- D:\python\app:/home/app
environment:
PYTHONUNBUFFERED: 'true'
DATABASE_URL: mysql://python:python_123456@mysql:3306/python_db
REDIS_URL: redis://redis:6379/0
networks:
- app_network
nginx:
build:
context: ./nginx
container_name: nginx
ports:
- "5000:5000" # Nginx's internal port 5000 mapped to host port 5000
depends_on:
- flask
volumes:
- ./nginx/conf:/etc/nginx/conf.d
networks:
- app_network
# 容器mysql
mysql:
build:
context: ./mysql
container_name: mysql
ports:
- "3307:3306" # MySQL's internal port 3306 mapped to host port 3307
volumes:
- ./mysql/data:/var/lib/mysql
- ./mysql/conf:/etc/mysql/conf.d
environment:
MYSQL_ROOT_PASSWORD: 123456
MYSQL_DATABASE: python_db
MYSQL_USER: python
MYSQL_PASSWORD: python_123456
networks:
- app_network
redis:
build:
context: ./redis
container_name: redis
ports:
- "6380:6379" # Redis's internal port 6379 mapped to host port 6380
supervisord:
build:
context: ./supervisord
container_name: supervisord
ports:
- "9001:9001" # Supervisord's web interface port
volumes:
- ./supervisord/conf:/etc/supervisor/conf.d
depends_on:
- flask
- nginx
networks:
- app_network
networks:
app_network:
driver: bridge
二、使用步骤
1.安装拓展
代码如下(示例):
# Use Python 3.8.6 base image
FROM python:3.8.6
# Set the working directory in the container
WORKDIR /home/app
# Copy requirements file to the container
COPY requirements.txt /home/app/
RUN apt-get update && apt-get install -y lsof
# Install the dependencies
# RUN pip install --no-cache-dir -r requirements.txt
RUN pip install -i https://pypi.tuna.tsinghua.edu.cn/simple -r requirements.txt
# RUN pip install Flask Flask-MySQLdb mysqlclient flask-login bcrypt Flask-SQLAlchemy PyMySQ
RUN pip install Flask Flask-MySQLdb mysqlclient flask-login bcrypt Flask-SQLAlchemy PyMySQL Flask-Migrate
# RUN pip install flask-restful
# 跨域访问
# RUN pip install flask-cors
# JWT
RUN pip install flask-jwt-extended
# SocketIO
RUN pip install eventlet flask-socketio
# requests
RUN pip install requests
# Copy the application code to the container
COPY . /home/app/
# Expose the Flask port
EXPOSE 5000
# Start the Flask application with Gunicorn
CMD ["gunicorn", "--bind", "0.0.0.0:5000", "--worker-class", "gthread", "--threads", "4", "wsgi:app"]
2.flask配置mysql连接
代码如下(示例):
from flask import Flask
from flask_login import LoginManager
# from .models import init_app as init_models
from .models import db
# 这里使用views.init_app()方法初始化视图蓝图,也可以把文件夹名称改成其他比如controller/api之类的
from .views import init_views
# from flask_mysqldb import MySQL
from flask_sqlalchemy import SQLAlchemy
#引入日志
import logging
from logging.handlers import RotatingFileHandler
# 引入数据库迁移扩展
from flask_migrate import Migrate
# 引入 restful 扩展
# from flask_restful import Api, Resource, reqparse
# 引入 JWT 扩展
from flask_jwt_extended import JWTManager, create_access_token, jwt_required
# mysql = MySQL()
# 初始化 SQLAlchemy
db = SQLAlchemy()
def create_app():
app = Flask(__name__)
# 初始化 restful 扩展
# api = Api(app, default_mediatype="application/json")
app.config['SECRET_KEY'] = 'test' # 用于加密会话数据
# 配置 MySQL 数据库连接 -- mysql+pymysql://用户名:密码@ip:端口/数据库名称
# app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://python:[email protected]:3306/python_db'
app.config['SQLALCHEMY_DATABASE_URI'] = 'mysql+pymysql://python:python_123456@mysql:3306/python_db'
# 禁用 SQLAlchemy 的修改追踪功能,这样可以提高性能
app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
# 初始化 db
db.init_app(app)
# 初始化数据库迁移扩展
migrate = Migrate(app, db)
# 初始化 JWT 扩展
jwt = JWTManager(app)
# login_manager = LoginManager()
# login_manager.init_app(app)
# login_manager.login_view = 'index.login' # 定义登录视图的端点
# 注册视图蓝图
init_views(app)
# 配置日志记录器
if not app.debug:
handler = RotatingFileHandler('app.log', maxBytes=1000000, encoding='utf-8', backupCount=3)
handler.setLevel(logging.INFO)
formatter = logging.Formatter(
'%(asctime)s %(levelname)s: %(message)s [in %(pathname)s:%(lineno)d]'
)
handler.setFormatter(formatter)
app.logger.addHandler(handler)
app.logger.setLevel(logging.DEBUG)
return app
3.简单使用
视图:
# 查询所有用户
@index_blueprint.route('/users', methods=['GET'])
#jwt修饰器,代表该接口需要使用token验证 格式 - Authorization:Bearer
@jwt_required()
def get_all_users():
logging.debug('This is a debug message')
try:
users = User.get_all_users()
user_dicts = [user.to_dict() for user in users]
return jsonify(user_dicts), 200
except Exception as e:
logging.error(f"Error fetching users: {e}")
return jsonify({"error": "An error occurred"}), 500
# 创建新用户
@index_blueprint.route('/create_user', methods=['POST'])
def create_user():
data = request.get_json()
username = data.get('username')
password = data.get('password')
try:
User.add_user(username, password)
return jsonify({'message': 'User created successfully!'}), 201
except Exception as e:
logging.error(f"Error creating user: {e}")
abort(500)
模型:
from app import db
import logging
class User(db.Model):
# __tablename__ = 'user'
id = db.Column(db.Integer, primary_key=True)
username = db.Column(db.String(255), unique=True, nullable=False)
password = db.Column(db.String(255), nullable=False)
def __init__(self, username, password):
self.username = username
self.password = password
def __repr__(self):
return f'<User {self.username}>'
def to_dict(self):
return {
'id': self.id,
'username': self.username
}
@staticmethod
def get_all_users():
return User.query.all()
@staticmethod
def add_user(username, password):
new_user = User(username=username, password=password)
db.session.add(new_user)
db.session.commit()
return new_user
蓝图注册:
def init_views(app):
# 注册index蓝图
from .index import index_blueprint
app.register_blueprint(index_blueprint)
标签:flask,app,db,更新,python,mysql,import,docker
From: https://blog.csdn.net/weixin_42695345/article/details/142071170