文章

Flask - 環境變量與配置管理

發布於
作者 Ned Hsu 5 分鐘閱讀

目標

  • 安裝並配置 python-dotenv
  • 重構配置以使用環境變量
  • 在不同環境中測試應用

步驟

  1. 準備環境

    • 繼續使用 flask_api/ 項目結構，激活虛擬環境：
      1
2

      # Windows: flask_api_env\Scripts\activate
# macOS/Linux: source flask_api_env/bin/activate
    • 安裝 python-dotenv：
      1

      pip install python-dotenv

  2. 創建環境變量文件

    • 在項目根目錄下創建 .env 文件：
      1
2
3
4
5
6

      FLASK_ENV=development
SECRET_KEY=your-secret-key-here
DATABASE_URL=sqlite:///blog.db
REDIS_URL=redis://localhost:6379/0
UPLOAD_FOLDER=uploads
MAX_CONTENT_LENGTH=16777216  # 16MB
    • 更新 .gitignore，確保不提交 .env：
      1

      .env

  3. 重構配置

    • 修改 app/config.py，使用環境變量：

      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

      import os
from dotenv import load_dotenv

# 加載 .env 文件
load_dotenv()

class Config:
    SECRET_KEY = os.getenv('SECRET_KEY', 'default-secret-key')
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    SQLALCHEMY_DATABASE_URI = os.getenv('DATABASE_URL')
    CACHE_TYPE = 'simple'
    CACHE_DEFAULT_TIMEOUT = 300
    UPLOAD_FOLDER = os.getenv('UPLOAD_FOLDER')
    MAX_CONTENT_LENGTH = int(os.getenv('MAX_CONTENT_LENGTH', 16 * 1024 * 1024))

class DevelopmentConfig(Config):
    DEBUG = True

class ProductionConfig(Config):
    DEBUG = False

class TestingConfig(Config):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = 'sqlite:///test.db'

config_map = {
    'development': DevelopmentConfig,
    'production': ProductionConfig,
    'testing': TestingConfig
}

  4. 更新應用初始化

    • 修改 app/init.py，整合環境變量：

      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

      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])

    # Heroku 特定配置
    if os.getenv('DATABASE_URL'):
        app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv('DATABASE_URL').replace('postgres://', 'postgresql://')

    # 確保上傳目錄存在
    os.makedirs(app.config['UPLOAD_FOLDER'], exist_ok=True)
    app.static_folder = app.root_path

    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

  5. 運行本地測試

    • 開發環境：
      1

      python run.py
      • 檢查是否使用 .env 中的配置（例如 SECRET_KEY）。
    • 生產環境模擬：
      1
2

      export FLASK_ENV=production
python run.py
      • 確認 DEBUG=False。

  6. 更新部署

    • 更新 requirements.txt：
      1

      pip freeze > requirements.txt
    • 提交更改並推送：
      1
2
3

      git add .
git commit -m "Add environment variable support"
git push heroku main
    • 在 Heroku 上設置環境變量：
      1
2
3
4
5
6

      heroku config:set FLASK_ENV=production
heroku config:set SECRET_KEY=your-very-secret-key
heroku config:set DATABASE_URL=<heroku-postgres-url>
heroku config:set REDIS_URL=<heroku-redis-url>
heroku config:set UPLOAD_FOLDER=uploads
heroku config:set MAX_CONTENT_LENGTH=16777216

  7. 測試應用

    • 本地：
      • POST /api/v1/users：應正常工作。
      • GET /api/v1/posts：確認緩存和日誌。
    • Heroku：
      • 訪問 https://my-blog-api.herokuapp.com/api/v1/posts。
      • 使用 Postman 測試文件上傳（如果已禁用，應返回無 image_path）。

  8. 作業

    • 添加一個 .env.test 文件，用於測試環境，並修改測試腳本加載它。
    • 在 config.py 中添加自定義配置項（例如 MAIL_SERVER），並在應用中使用。

注意事項

  • .env 文件不應提交到版本控制，生產環境使用平台提供的環境變量。
  • 確保 SECRET_KEY 在生產中是唯一的且安全的。
  • Heroku 的 DATABASE_URL 和 REDIS_URL 會自動設置，優先級高於 .env。
Software, Python
Flask
本文章以 CC BY 4.0 授權