文章

Flask - 部署

目標

  • 配置 Gunicorn 作為 WSGI 服務器
  • 準備應用並部署到 Heroku
  • 測試線上 API

步驟

  1. 準備環境

    • 繼續使用 flask_api/ 項目結構,激活虛擬環境:
      1
      2
      
      # Windows: flask_api_env\Scripts\activate
      # macOS/Linux: source flask_api_env/bin/activate
      
    • 安裝 Gunicorn 和 Heroku CLI:
      1
      
      pip install gunicorn
      
  2. 本地測試 Gunicorn

    • 在項目根目錄下運行:
      1
      
      gunicorn -w 4 -b 0.0.0.0:5000 run:app
      
      • -w 4:啟動 4 個工作進程。
      • -b 0.0.0.0:5000:綁定到 5000 端口。
      • run:app:指定應用入口(run.py 中的 app)。
    • 訪問 http://localhost:5000/api/v1/posts,確保正常工作。
  3. 準備部署文件

    • requirements.txt:生成依賴列表:
      1
      
      pip freeze > requirements.txt
      
    • Procfile:告訴 Heroku 如何運行應用,新建 Procfile(無擴展名):
      1
      2
      
      web: gunicorn run:app
      worker: celery -A app worker --loglevel=info
      
    • .gitignore:忽略不必要的文件:
      1
      2
      3
      4
      5
      6
      7
      8
      9
      
      __pycache__/
      *.pyc
      *.pyo
      *.pyd
      *.db
      *.log
      flask_api_env/
      uploads/
      .env
      
    • 修改 app/init.py,適應 Heroku 環境:

      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
      29
      30
      31
      32
      33
      34
      35
      36
      37
      38
      39
      40
      41
      42
      43
      44
      45
      46
      47
      48
      49
      50
      51
      52
      53
      54
      55
      56
      57
      58
      59
      60
      61
      62
      63
      64
      65
      66
      67
      68
      69
      70
      71
      72
      73
      74
      75
      76
      77
      78
      79
      80
      81
      82
      83
      84
      85
      86
      87
      88
      89
      90
      91
      92
      93
      94
      95
      96
      97
      98
      99
      100
      101
      102
      103
      104
      105
      106
      107
      108
      109
      110
      111
      112
      113
      114
      115
      116
      117
      118
      119
      120
      121
      122
      123
      124
      125
      126
      127
      128
      129
      130
      131
      132
      133
      134
      135
      136
      137
      138
      139
      140
      141
      142
      143
      144
      145
      
      from flask import Flask, jsonify, g
      from flask_sqlalchemy import SQLAlchemy
      from flask_marshmallow import Marshmallow
      from flask_bcrypt import Bcrypt
      from flask_restx import Api
      from flask_caching import Cache
      from flask_limiter import Limiter
      from flask_limiter.util import get_remote_address
      from flask_cors import CORS
      import jwt
      from functools import wraps
      from .routes.v1.todos import todos_bp as todos_v1_bp
      from .routes.v1.users import users_bp as users_v1_bp
      from .routes.v1.posts import posts_bp as posts_v1_bp
      from .routes.v2.todos import todos_bp as todos_v2_bp
      from .routes.v2.posts import posts_bp as posts_v2_bp
      from .config import config_map
      from .celery_config import make_celery
      import os
      import logging
      from logging.handlers import RotatingFileHandler
      
      db = SQLAlchemy()
      ma = Marshmallow()
      bcrypt = Bcrypt()
      cache = Cache()
      limiter = Limiter(key_func=get_remote_address)
      
      def setup_logging(app):
          if not app.debug:
              handler = RotatingFileHandler('app.log', maxBytes=10000, 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)
          console_handler = logging.StreamHandler()
          console_handler.setLevel(logging.DEBUG)
          console_handler.setFormatter(logging.Formatter('%(asctime)s %(levelname)s: %(message)s'))
          app.logger.addHandler(console_handler)
          app.logger.setLevel(logging.DEBUG)
      
      def create_app():
          app = Flask(__name__)
          env = os.getenv('FLASK_ENV', 'development')
          app.config.from_object(config_map[env])
          app.config['CACHE_TYPE'] = 'simple'
          app.config['CACHE_DEFAULT_TIMEOUT'] = 300
          app.config['UPLOAD_FOLDER'] = os.path.join(app.root_path, 'uploads')
          app.config['MAX_CONTENT_LENGTH'] = 16 * 1024 * 1024
          os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
          app.static_folder = app.root_path
      
          # Heroku 環境配置
          app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL', app.config['SQLALCHEMY_DATABASE_URI'])
          if app.config['SQLALCHEMY_DATABASE_URI'].startswith('postgres://'):
              app.config['SQLALCHEMY_DATABASE_URI'] = app.config['SQLALCHEMY_DATABASE_URI'].replace('postgres://', 'postgresql://')
      
          db.init_app(app)
          ma.init_app(app)
          bcrypt.init_app(app)
          cache.init_app(app)
          limiter.init_app(app)
          setup_logging(app)
      
          CORS(app, resources={r"/api/*": {"origins": "*"}})
      
          api = Api(app,
                    title='Blog API',
                    version='1.0',
                    description='A simple blog API with user and post management',
                    doc='/api/docs/',
                    authorizations={
                        'jwt': {
                            'type': 'apiKey',
                            'in': 'header',
                            'name': 'Authorization',
                            'description': 'Enter "Bearer <token>"'
                        }
                    })
      
          global celery
          celery = make_celery(app)
          celery.conf.broker_url = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
          celery.conf.result_backend = os.getenv('REDIS_URL', 'redis://localhost:6379/0')
      
          app.register_blueprint(todos_v1_bp, url_prefix='/api/v1')
          app.register_blueprint(users_v1_bp, url_prefix='/api/v1')
          app.register_blueprint(posts_v1_bp, url_prefix='/api/v1')
          app.register_blueprint(todos_v2_bp, url_prefix='/api/v2')
          app.register_blueprint(posts_v2_bp, url_prefix='/api/v2')
      
          @app.errorhandler(404)
          def not_found(error):
              app.logger.error(f'404 error: {str(error)}')
              return jsonify({'error': 'Not Found', 'message': str(error)}), 404
      
          @app.errorhandler(400)
          def bad_request(error):
              app.logger.warning(f'400 error: {str(error)}')
              return jsonify({'error': 'Bad Request', 'message': str(error)}), 400
      
          @app.errorhandler(500)
          def internal_error(error):
              app.logger.critical(f'500 error: {str(error)}')
              return jsonify({'error': 'Internal Server Error', 'message': 'Something went wrong on our end'}), 500
      
          with app.app_context():
              db.create_all()
      
          return app
      
      def login_required(f):
          @wraps(f)
          def decorated_function(*args, **kwargs):
              from .models import User
              token = request.headers.get('Authorization')
              if not token:
                  abort(401, description='Missing token')
              try:
                  if token.startswith('Bearer '):
                      token = token[7:]
                  data = jwt.decode(token, app.config['SECRET_KEY'], algorithms=['HS256'])
                  user = User.query.get(data['user_id'])
                  if not user:
                      abort(401, description='Invalid token')
                  g.current_user = user
              except jwt.ExpiredSignatureError:
                  abort(401, description='Token has expired')
              except jwt.InvalidTokenError:
                  abort(401, description='Invalid token')
              return f(*args, **kwargs)
          return decorated_function
      
      def admin_required(f):
          @wraps(f)
          @login_required
          def decorated_function(*args, **kwargs):
              if not g.current_user.is_admin:
                  abort(403, description='Admin access required')
              return f(*args, **kwargs)
          return decorated_function
      
      celery = None
      
  4. 部署到 Heroku

    • 初始化 Git 倉庫
      1
      2
      3
      
      git init
      git add .
      git commit -m "Initial commit"
      
    • 創建 Heroku 應用
      1
      
      heroku create my-blog-api
      
    • 添加 Heroku 插件
      • PostgreSQL(數據庫):
        1
        
        heroku addons:create heroku-postgresql:hobby-dev -a my-blog-api
        
      • Redis(Celery 和緩存):
        1
        
        heroku addons:create heroku-redis:hobby-dev -a my-blog-api
        
    • 設置環境變量
      1
      2
      
      heroku config:set FLASK_ENV=production
      heroku config:set SECRET_KEY=your-very-secret-key
      
    • 推送代碼
      1
      
      git push heroku main
      
    • 初始化數據庫
      1
      
      heroku run "flask db upgrade"  # 如果使用 Flask-Migrate
      
      • 如果未使用 Migrate,手動運行:
        1
        
        heroku run python -c "from app import db, create_app; app = create_app(); with app.app_context(): db.create_all()"
        
  5. 測試線上 API

    • 獲取應用 URL:
      1
      
      heroku open
      
    • 使用 Postman 測試:
      • GET https://my-blog-api.herokuapp.com/api/v1/posts
      • POST https://my-blog-api.herokuapp.com/api/v1/users
        • Body:{"username": "alice", "password": "1234"}
    • 檢查文檔:https://my-blog-api.herokuapp.com/api/docs/
  6. 處理文件上傳

    • Heroku 的文件系統是臨時的,需使用雲存儲(如 AWS S3)。簡單修改 app/routes/v1/posts.py,禁用文件上傳:
      1
      2
      3
      4
      5
      6
      
      # 在 post 方法中註釋掉圖片保存部分
      if image and allowed_file(image.filename) and os.getenv('FLASK_ENV') != 'production':
          filename = f"{uuid.uuid4().hex}.{image.filename.rsplit('.', 1)[1].lower()}"
          image.save(os.path.join(api.app.config['UPLOAD_FOLDER'], filename))
          post.image_path = f"/uploads/{filename}"
          api.app.logger.info(f'Image uploaded: {filename}')
      
  7. 作業

    • 集成 AWS S3 處理文件上傳(提示:使用 boto3)。
    • 添加環境變量 REDIS_URL 到緩存和 Limiter 配置。

注意事項

  • Heroku 免費層有休眠機制,啟動可能稍慢。
  • 確保 SECRET_KEY 安全且唯一。
  • Celery 在 Heroku 上需啟動 worker,檢查日誌:
    1
    
    heroku logs --tail
    
  • 生產環境應使用域名和 HTTPS(Heroku 默認提供)。

本文章以 CC BY 4.0 授權