Feature/app factory pattern

api/vercel_function.py

@@ -1 +1,5 @@
-from app import app
+from app import create_app
+from config import ProductionConfig
+
+
+app = create_app(ProductionConfig)

app/__init__.py

@@ -1,99 +1,28 @@
-import os
-from datetime import timedelta
+from flask import Flask
 
-from flask import Flask, jsonify
-from flask_jwt_extended import JWTManager
-from flask_migrate import Migrate
-from flask_sqlalchemy import SQLAlchemy
-from dotenv import load_dotenv
-from sqlalchemy import MetaData
-from flask_smorest import Api
+from app.extensions import api, db, jwt, migrate
+from config import DevelopmentConfig
 
 
-def register_blueprints():
+def create_app(config_class=DevelopmentConfig):
+    app = Flask(__name__)
+    app.config.from_object(config_class)
+
+    # initialize extenstions
+    db.init_app(app)
+    migrate.init_app(app)
+    jwt.init_app(app)
+    api.init_app(app)
+
+    # register blueprints
+    from app.routes.auth import bp as auth_bp
     from app.routes.category import bp as category_bp
-    from app.routes.subcategory import bp as subcategory_bp
     from app.routes.product import bp as product_bp
-    from app.routes.auth import bp as auth_bp
+    from app.routes.subcategory import bp as subcategory_bp
 
     api.register_blueprint(category_bp, url_prefix="/categories")
     api.register_blueprint(subcategory_bp, url_prefix="/subcategories")
     api.register_blueprint(product_bp, url_prefix="/products")
     api.register_blueprint(auth_bp, url_prefix="/auth")
 
-
-app = Flask(__name__)
-
-load_dotenv()
-
-# sqlalchemy
-app.config['SQLALCHEMY_DATABASE_URI'] = os.getenv("SQLALCHEMY_DATABASE_URI")
-app.config['SQLALCHEMY_TRACK_MODIFICATIONS'] = False
-
-# jwt
-app.config["JWT_SECRET_KEY"] = os.getenv("JWT_SECRET_KEY")
-app.config["JWT_ACCESS_TOKEN_EXPIRES"] = timedelta(hours=3)
-app.config["JWT_REFRESH_TOKEN_EXPIRES"] = timedelta(days=3)
-
-# flask-smorest
-app.config["API_TITLE"] = "Ecommerce REST API"
-app.config["API_VERSION"] = "v1"
-app.config["OPENAPI_VERSION"] = "3.0.2"
-
-# flask-smorest openapi swagger
-app.config["OPENAPI_URL_PREFIX"] = "/"
-app.config["OPENAPI_SWAGGER_UI_PATH"] = "/"
-app.config["OPENAPI_SWAGGER_UI_URL"] = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"
-
-# flask-smorest Swagger UI top level authorize dialog box
-app.config["API_SPEC_OPTIONS"] = {
-    "components": {
-        "securitySchemes": {
-            "access_token": {
-                "type": "http",
-                "scheme": "bearer",
-                "bearerFormat": "JWT",
-                "description": "Enter your JWT access token",
-            },
-            "refresh_token": {
-                "type": "http",
-                "scheme": "bearer",
-                "bearerFormat": "JWT",
-                "description": "Enter your JWT refresh token",
-            },
-        }
-    }
-}
-
-# PostgreSQL-compatible naming convention (to follow the naming convention already used in the DB)
-# https://stackoverflow.com/questions/4107915/postgresql-default-constraint-names
-naming_convention = {
-    "ix": "%(table_name)s_%(column_0_name)s_idx",       # Indexes
-    "uq": "%(table_name)s_%(column_0_name)s_key",       # Unique constraints
-    "ck": "%(table_name)s_%(constraint_name)s_check",   # Check constraints
-    "fk": "%(table_name)s_%(column_0_name)s_fkey",      # Foreign keys
-    "pk": "%(table_name)s_pkey"                         # Primary keys
-}
-metadata = MetaData(naming_convention=naming_convention)
-db = SQLAlchemy(app, metadata=metadata)
-migrate = Migrate(app, db)
-jwt = JWTManager(app)
-api = Api(app)
-
-register_blueprints()
-
-
-@jwt.expired_token_loader
-def expired_token_callback(jwt_header, jwt_payload):
-    err = "Access token expired. Use your refresh token to get a new one."
-    if jwt_payload['type'] == 'refresh':
-        err = "Refresh token expired. Please login again."
-    return jsonify(code="token_expired", error=err), 401
-
-@jwt.invalid_token_loader
-def invalid_token_callback(error):
-    return jsonify(code="invalid_token", error="Invalid token provided."), 401
-
-@jwt.unauthorized_loader
-def missing_token_callback(error):
-    return jsonify(code="authorization_required", error="JWT needed for this operation. Login, if needed."), 401
+    return app

app/extensions.py

@@ -0,0 +1,43 @@
+from flask import jsonify
+from flask_jwt_extended import JWTManager
+from flask_migrate import Migrate
+from flask_smorest import Api
+from flask_sqlalchemy import SQLAlchemy
+from sqlalchemy import MetaData
+
+
+# PostgreSQL-compatible naming convention (to follow the naming convention already used in the DB)
+# https://stackoverflow.com/questions/4107915/postgresql-default-constraint-names
+naming_convention = {
+    "ix": "%(table_name)s_%(column_0_name)s_idx",  # Indexes
+    "uq": "%(table_name)s_%(column_0_name)s_key",  # Unique constraints
+    "ck": "%(table_name)s_%(constraint_name)s_check",  # Check constraints
+    "fk": "%(table_name)s_%(column_0_name)s_fkey",  # Foreign keys
+    "pk": "%(table_name)s_pkey",  # Primary keys
+}
+metadata = MetaData(naming_convention=naming_convention)
+db = SQLAlchemy(metadata=metadata)
+migrate = Migrate(db)
+jwt = JWTManager()
+api = Api()
+
+
+@jwt.expired_token_loader
+def expired_token_callback(jwt_header, jwt_payload):
+    err = "Access token expired. Use your refresh token to get a new one."
+    if jwt_payload["type"] == "refresh":
+        err = "Refresh token expired. Please login again."
+    return jsonify(code="token_expired", error=err), 401
+
+
+@jwt.invalid_token_loader
+def invalid_token_callback(error):
+    return jsonify(code="invalid_token", error="Invalid token provided."), 401
+
+
+@jwt.unauthorized_loader
+def missing_token_callback(error):
+    return jsonify(
+        code="authorization_required",
+        error="JWT needed for this operation. Login, if needed.",
+    ), 401

config.py

@@ -0,0 +1,62 @@
+import os
+from datetime import timedelta
+
+from dotenv import load_dotenv
+
+
+load_dotenv()
+
+
+class Config:
+    # sqlalchemy
+    SQLALCHEMY_TRACK_MODIFICATIONS = False
+
+    # jwt
+    JWT_SECRET_KEY = os.getenv("JWT_SECRET_KEY")
+    JWT_ACCESS_TOKEN_EXPIRES = timedelta(hours=3)
+    JWT_REFRESH_TOKEN_EXPIRES = timedelta(days=3)
+
+    # flask-smorest
+    API_TITLE = "Ecommerce REST API"
+    API_VERSION = "v1"
+    OPENAPI_VERSION = "3.0.2"
+
+    # flask-smorest openapi swagger
+    OPENAPI_URL_PREFIX = "/"
+    OPENAPI_SWAGGER_UI_PATH = "/"
+    OPENAPI_SWAGGER_UI_URL = "https://cdn.jsdelivr.net/npm/swagger-ui-dist/"
+
+    # flask-smorest Swagger UI top level authorize dialog box
+    API_SPEC_OPTIONS = {
+        "components": {
+            "securitySchemes": {
+                "access_token": {
+                    "type": "http",
+                    "scheme": "bearer",
+                    "bearerFormat": "JWT",
+                    "description": "Enter your JWT access token",
+                },
+                "refresh_token": {
+                    "type": "http",
+                    "scheme": "bearer",
+                    "bearerFormat": "JWT",
+                    "description": "Enter your JWT refresh token",
+                },
+            }
+        }
+    }
+
+
+class DevelopmentConfig(Config):
+    DEBUG = True
+    SQLALCHEMY_DATABASE_URI = os.getenv("SQLALCHEMY_DATABASE_URI")
+
+
+class TestingConfig(Config):
+    TESTING = True
+    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"
+    JWT_SECRET_KEY = os.urandom(24).hex()
+
+
+class ProductionConfig(Config):
+    SQLALCHEMY_DATABASE_URI = os.getenv("SQLALCHEMY_DATABASE_URI")

populate_db.py

@@ -1,11 +1,13 @@
 from faker import Faker
-from app import app, db
+from app import create_app, db
 from app.models import Category, Subcategory, Product, category_subcategory, subcategory_product
 import random
 
 
+app = create_app()
 fake = Faker()
 
+
 def create_categories(num=5):
     categories = []
     for _ in range(num):

tests/conftest.py

@@ -1,24 +1,30 @@
-import os
-
 import pytest
 
-# TODO: Fix hack. Changes the env var before initializing the db for testing
-os.environ["SQLALCHEMY_DATABASE_URI"] = "sqlite:///:memory:"
-os.environ["JWT_SECRET_KEY"] = os.urandom(24).hex()
-
-from app import app, db
+from app import create_app, db
+from config import TestingConfig
 from tests import utils
 
 
 @pytest.fixture
-def client():
-    app.config["TESTING"] = True
-    with app.test_client() as client:
-        with app.app_context():
-            db.create_all()
-        yield client
-        with app.app_context():
-            db.drop_all()
+def app():
+    app = create_app(TestingConfig)
+
+    # setup
+    app_context = app.app_context()
+    app_context.push()
+    db.create_all()
+
+    yield app
+
+    # teardown
+    db.session.remove()
+    db.drop_all()
+    app_context.pop()
+
+
+@pytest.fixture
+def client(app):
+    return app.test_client()
 
 
 @pytest.fixture

tests/test_auth.py

@@ -16,22 +16,19 @@ class TestAuth:
     @pytest.fixture(autouse=True)
     def setup(self, client):
         self.client = client
-        with client.application.app_context():
-            assert User.query.count() == 0
+        assert User.query.count() == 0
 
     def _verify_user_in_db(self, email, should_exist=True):
-        with self.client.application.app_context():
-            user = User.get(email=email)
-            if should_exist:
-                assert user is not None
-                assert user.email == email
-                return user
-            else:
-                assert user is None
+        user = User.get(email=email)
+        if should_exist:
+            assert user is not None
+            assert user.email == email
+            return user
+        else:
+            assert user is None
 
     def _count_users(self):
-        with self.client.application.app_context():
-            return User.query.count()
+        return User.query.count()
 
     def _test_invalid_request_data(self, endpoint, expected_status=422):
         response = self.client.post(endpoint, json={})
@@ -47,9 +44,7 @@ def _test_invalid_request_data(self, endpoint, expected_status=422):
         assert response.status_code == expected_status
 
     def _decode_token(self, token):
-        # Needs Flask app context for secret/algorithms from current_app.config
-        with self.client.application.app_context():
-            return decode_token(token, allow_expired=False)
+        return decode_token(token, allow_expired=False)
 
     def _assert_jwt_structure(self, token, expected_sub, expected_type, fresh=False):
         assert token.count(".") == 2, f"Token does not have three segments: {token}"

tests/test_category.py

@@ -13,22 +13,19 @@ class TestCategory:
     @pytest.fixture(autouse=True)
     def setup(self, client):
         self.client = client
-        with client.application.app_context():
-            assert Category.query.count() == 0
+        assert Category.query.count() == 0
 
     def _count_categories(self):
-        with self.client.application.app_context():
-            return Category.query.count()
+        return Category.query.count()
 
     def _verify_category_in_db(self, name, should_exist=True):
-        with self.client.application.app_context():
-            category = Category.query.filter_by(name=name).first()
-            if should_exist:
-                assert category is not None
-                assert category.name == name
-                return category
-            else:
-                assert category is None
+        category = Category.query.filter_by(name=name).first()
+        if should_exist:
+            assert category is not None
+            assert category.name == name
+            return category
+        else:
+            assert category is None
 
     def test_create_category(self, create_category):
         response = create_category(self.TEST_CATEGORY_NAME)

tests/test_product.py

@@ -14,22 +14,19 @@ class TestProduct:
     @pytest.fixture(autouse=True)
     def setup(self, client):
         self.client = client
-        with client.application.app_context():
-            assert Product.query.count() == 0
+        assert Product.query.count() == 0
 
     def _count_products(self):
-        with self.client.application.app_context():
-            return Product.query.count()
+        return Product.query.count()
 
     def _verify_product_in_db(self, name, should_exist=True):
-        with self.client.application.app_context():
-            product = Product.query.filter_by(name=name).first()
-            if should_exist:
-                assert product is not None
-                assert product.name == name
-                return product
-            else:
-                assert product is None
+        product = Product.query.filter_by(name=name).first()
+        if should_exist:
+            assert product is not None
+            assert product.name == name
+            return product
+        else:
+            assert product is None
 
     def test_create_product(self, create_product):
         response = create_product(self.TEST_PRODUCT_NAME, self.TEST_PRODUCT_DESC)