Skip to main content

​
Overview

The Python Flask service is a RESTful API microservice that handles core e-commerce functionality including user authentication, product management, order processing, and payment calculations. It uses Flask with SQLAlchemy ORM and PostgreSQL for production environments.

​
Technology Stack

  • Framework: Flask 2.3+
  • ORM: SQLAlchemy 3.1
  • Database: PostgreSQL (SQLite for testing)
  • Authentication: JWT (Flask-JWT-Extended)
  • Password Hashing: bcrypt
  • CORS: Flask-CORS
  • Testing: pytest with pytest-flask

​
Setup and Installation

1

Clone and navigate to the service

cd python-service
2

Install dependencies

pip install -r requirements.txt
3

Configure environment

Create a .env file or set environment variables:
export FLASK_ENV=development
export SECRET_KEY=your-secret-key
export DATABASE_USER=appuser
export DATABASE_PASSWORD=apppassword
export DATABASE_NAME=ecommerce
export DATABASE_PORT=5432
4

Run the service

python run.py
The service will start on http://localhost:5000

​
Project Structure

python-service/
├── app/
│   ├── __init__.py          # Application factory
│   ├── config.py            # Configuration classes
│   ├── models/              # Database models
│   │   ├── user.py
│   │   ├── product.py
│   │   └── order.py
│   ├── routes/              # API endpoints
│   │   ├── auth.py
│   │   ├── products.py
│   │   ├── orders.py
│   │   └── payments.py
│   └── services/
│       └── payment_service.py
├── tests/
│   ├── conftest.py          # Test fixtures
│   ├── test_auth.py
│   ├── test_products.py
│   └── test_orders.py
├── requirements.txt
└── run.py                   # Application entry point

​
Configuration

The service supports multiple environments configured in app/config.py:1:
app/config.py
class BaseConfig:
    SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-key")
    SQLALCHEMY_TRACK_MODIFICATIONS = False
    JWT_SECRET_KEY = os.environ.get("SECRET_KEY", "dev-secret-key")
    JWT_ACCESS_TOKEN_EXPIRES = 3600


class DevelopmentConfig(BaseConfig):
    DEBUG = True
    SQLALCHEMY_DATABASE_URI = (
        f"postgresql://"
        f"{os.environ.get('DATABASE_USER', 'appuser')}:"
        f"{os.environ.get('DATABASE_PASSWORD', 'apppassword')}@"
        f"localhost:"
        f"{os.environ.get('DATABASE_PORT', '5432')}/"
        f"{os.environ.get('DATABASE_NAME', 'ecommerce')}"
    )


class ProductionConfig(BaseConfig):
    DEBUG = False
    SQLALCHEMY_DATABASE_URI = (
        f"postgresql://"
        f"{os.environ.get('DB_USER', 'appuser')}:"
        f"{os.environ.get('DB_PASS', 'apppassword')}@"
        f"{os.environ.get('DB_HOST', 'localhost')}:"
        f"{os.environ.get('DB_PORT', '5432')}/"
        f"{os.environ.get('DB_NAME', 'ecommerce')}"
    )


class TestingConfig(BaseConfig):
    TESTING = True
    SQLALCHEMY_DATABASE_URI = "sqlite:///:memory:"
Configuration is loaded based on the FLASK_ENV environment variable in app/config.py:61.

​
Database Models

​
User Model

Defined in app/models/user.py:5:
app/models/user.py
class User(db.Model):
    __tablename__ = "users"

    id = db.Column(db.Integer, primary_key=True)
    email = db.Column(db.String(255), unique=True, nullable=False, index=True)
    password_hash = db.Column(db.String(255), nullable=False)
    name = db.Column(db.String(255), nullable=False)
    role = db.Column(db.String(50), default="customer")
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    orders = db.relationship("Order", backref="user", lazy="dynamic")

    def to_dict(self):
        return {
            "id": self.id,
            "email": self.email,
            "name": self.name,
            "role": self.role,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }

​
Product Model

Defined in app/models/product.py:5:
app/models/product.py
class Product(db.Model):
    __tablename__ = "products"

    id = db.Column(db.Integer, primary_key=True)
    name = db.Column(db.String(255), nullable=False)
    description = db.Column(db.Text)
    price = db.Column(db.Numeric(10, 2), nullable=False)
    stock = db.Column(db.Integer, default=0)
    category = db.Column(db.String(100), index=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    def to_dict(self):
        return {
            "id": self.id,
            "name": self.name,
            "description": self.description,
            "price": float(self.price),
            "stock": self.stock,
            "category": self.category,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }

​
Order Model

Defined in app/models/order.py:5:
app/models/order.py
class Order(db.Model):
    __tablename__ = "orders"

    id = db.Column(db.Integer, primary_key=True)
    user_id = db.Column(db.Integer, db.ForeignKey("users.id"), nullable=False, index=True)
    status = db.Column(db.String(50), default="pending")
    subtotal = db.Column(db.Numeric(10, 2), nullable=False)
    tax = db.Column(db.Numeric(10, 2), nullable=False, default=0)
    discount_amount = db.Column(db.Numeric(10, 2), default=0)
    total = db.Column(db.Numeric(10, 2), nullable=False)
    discount_code = db.Column(db.String(50), nullable=True)
    created_at = db.Column(db.DateTime, default=datetime.utcnow)
    updated_at = db.Column(db.DateTime, default=datetime.utcnow, onupdate=datetime.utcnow)

    items = db.relationship("OrderItem", backref="order", lazy="select")

    def to_dict(self, include_items=False):
        result = {
            "id": self.id,
            "user_id": self.user_id,
            "status": self.status,
            "subtotal": float(self.subtotal),
            "tax": float(self.tax),
            "discount_amount": float(self.discount_amount),
            "total": float(self.total),
            "discount_code": self.discount_code,
            "created_at": self.created_at,
            "updated_at": self.updated_at,
        }
        if include_items:
            result["items"] = [item.to_dict() for item in self.items]
        return result

​
OrderItem Model

Defined in app/models/order.py:39:
app/models/order.py
class OrderItem(db.Model):
    __tablename__ = "order_items"

    id = db.Column(db.Integer, primary_key=True)
    order_id = db.Column(db.Integer, db.ForeignKey("orders.id"), nullable=False, index=True)
    product_id = db.Column(db.Integer, db.ForeignKey("products.id"), nullable=False)
    quantity = db.Column(db.Integer, nullable=False)
    unit_price = db.Column(db.Numeric(10, 2), nullable=False)
    total_price = db.Column(db.Numeric(10, 2), nullable=False)

    product = db.relationship("Product")

    def to_dict(self):
        return {
            "id": self.id,
            "order_id": self.order_id,
            "product_id": self.product_id,
            "product_name": self.product.name if self.product else None,
            "quantity": self.quantity,
            "unit_price": float(self.unit_price),
            "total_price": float(self.total_price),
        }

​
API Routes

​
Authentication Routes

Registered at /api/auth in app/__init__.py:47:

​
POST /api/auth/register

Register a new user with email, password, and name in app/routes/auth.py:11:
app/routes/auth.py
@auth_bp.route("/register", methods=["POST"])
def register():
    data = request.get_json()

    if not data or not data.get("email") or not data.get("password") or not data.get("name"):
        return jsonify({"error": "Missing required fields: email, password, name"}), 400

    if User.query.filter_by(email=data["email"]).first():
        return jsonify({"error": "Email already registered"}), 409

    password_hash = bcrypt.hashpw(
        data["password"].encode("utf-8"),
        bcrypt.gensalt()
    ).decode("utf-8")

    user = User(
        email=data["email"],
        password_hash=password_hash,
        name=data["name"],
    )
    db.session.add(user)
    db.session.commit()

    access_token = create_access_token(identity=str(user.id))
    return jsonify({
        "message": "User registered successfully",
        "user": user.to_dict(),
        "access_token": access_token,
    }), 201

​
POST /api/auth/login

Authenticate user and return JWT token in app/routes/auth.py:42:
app/routes/auth.py
@auth_bp.route("/login", methods=["POST"])
def login():
    data = request.get_json()

    if not data or not data.get("email") or not data.get("password"):
        return jsonify({"error": "Missing required fields: email, password"}), 400

    user = User.query.filter_by(email=data["email"]).first()
    if not user:
        return jsonify({"error": "Invalid email or password"}), 401

    is_valid = bcrypt.checkpw(
        data["password"].encode("utf-8"),
        user.password_hash
    )

    if not is_valid:
        return jsonify({"error": "Invalid email or password"}), 401

    access_token = create_access_token(identity=str(user.id))
    return jsonify({
        "message": "Login successful",
        "user": user.to_dict(),
        "access_token": access_token,
    }), 200

​
GET /api/auth/me

Get current authenticated user (requires JWT) in app/routes/auth.py:69.

​
Product Routes

Registered at /api/products in app/__init__.py:48:

​
GET /api/products

List products with pagination and optional category filter in app/routes/products.py:10:
app/routes/products.py
@products_bp.route("/", methods=["GET"])
def list_products():
    page = request.args.get("page", 1, type=int)
    per_page = request.args.get("per_page", 20, type=int)
    category = request.args.get("category")

    query = Product.query
    if category:
        query = query.filter_by(category=category)

    pagination = query.order_by(Product.created_at.desc()).paginate(
        page=page, per_page=per_page, error_out=False
    )

    return jsonify({
        "products": [p.to_dict() for p in pagination.items],
        "total": pagination.total,
        "page": page,
        "per_page": per_page,
        "pages": pagination.pages,
    }), 200

​
GET /api/products/:id

Get a single product by ID in app/routes/products.py:33.

​
POST /api/products

Create a new product (requires JWT) in app/routes/products.py:66.

​
GET /api/products/search

Search products by name or description in app/routes/products.py:42.

​
Order Routes

Registered at /api/orders in app/__init__.py:49:

​
GET /api/orders

List all orders for the authenticated user in app/routes/orders.py:11.

​
GET /api/orders/:id

Get a specific order with items in app/routes/orders.py:35.

​
POST /api/orders

Create a new order in app/routes/orders.py:47:
app/routes/orders.py
@orders_bp.route("/", methods=["POST"])
@jwt_required()
def create_order():
    user_id = get_jwt_identity()
    data = request.get_json()

    if not data or not data.get("items"):
        return jsonify({"error": "Order must include at least one item"}), 400

    subtotal = 0
    order_items = []

    for item_data in data["items"]:
        product = Product.query.get(item_data.get("product_id"))
        if not product:
            return jsonify({"error": f"Product {item_data.get('product_id')} not found"}), 404

        quantity = item_data.get("quantity", 1)

        if product.stock < quantity:
            return jsonify({
                "error": f"Insufficient stock for {product.name}. Available: {product.stock}"
            }), 400

        item_total = float(product.price) * quantity
        subtotal += item_total

        order_items.append({
            "product": product,
            "quantity": quantity,
            "unit_price": float(product.price),
            "total_price": item_total,
        })

    # Calculate tax and discount
    tax = calculate_tax(subtotal)
    discount_amount = 0
    discount_code = data.get("discount_code")

    if discount_code:
        subtotal, discount_amount = apply_discount(subtotal, discount_code)

    total = subtotal + tax - discount_amount

    order = Order(
        user_id=int(user_id),
        subtotal=subtotal,
        tax=tax,
        discount_amount=discount_amount,
        total=total,
        discount_code=discount_code,
    )
    db.session.add(order)
    db.session.flush()

    for item_data in order_items:
        order_item = OrderItem(
            order_id=order.id,
            product_id=item_data["product"].id,
            quantity=item_data["quantity"],
            unit_price=item_data["unit_price"],
            total_price=item_data["total_price"],
        )
        db.session.add(order_item)
        item_data["product"].stock -= item_data["quantity"]

    db.session.commit()

    return jsonify({"order": order.to_dict(include_items=True)}), 201

​
Payment Routes

Registered at /api/payments in app/__init__.py:50:

​
POST /api/payments/calculate

Calculate total with tax and discounts in app/routes/payments.py:11.

​
POST /api/payments/checkout

Process payment for an order in app/routes/payments.py:42.

​
Testing

The service uses pytest with fixtures for comprehensive testing.

​
Running Tests

pytest

# With coverage
pytest --cov=app

# Verbose output
pytest -v

​
Test Fixtures

Defined in tests/conftest.py:9:
tests/conftest.py
@pytest.fixture(scope="session")
def app():
    """Create application for testing."""
    app = create_app("testing")
    return app


@pytest.fixture(scope="function")
def db(app):
    """Create a fresh database for each test."""
    with app.app_context():
        _db.create_all()
        yield _db
        _db.session.rollback()
        _db.drop_all()


@pytest.fixture
def client(app, db):
    """Create a test client."""
    return app.test_client()


@pytest.fixture
def auth_token(client, sample_user):
    """Get a JWT token for the sample user."""
    from flask_jwt_extended import create_access_token
    from flask import current_app

    with current_app.app_context():
        token = create_access_token(identity=str(sample_user.id))
    return token

​
Example Test

From tests/test_auth.py:9:
tests/test_auth.py
class TestRegistration:
    """Tests for POST /api/auth/register."""

    def test_register_success(self, client):
        """Should register a new user successfully."""
        response = client.post(
            "/api/auth/register",
            data=json.dumps({
                "email": "[email protected]",
                "password": "securepassword123",
                "name": "New User",
            }),
            content_type="application/json",
        )
        assert response.status_code == 201
        data = response.get_json()
        assert data["user"]["email"] == "[email protected]"
        assert "access_token" in data

    def test_register_duplicate_email(self, client, sample_user):
        """Should return 409 when email is already registered."""
        response = client.post(
            "/api/auth/register",
            data=json.dumps({
                "email": "[email protected]",
                "password": "password123",
                "name": "Duplicate User",
            }),
            content_type="application/json",
        )
        assert response.status_code == 409

​
Application Factory

The service uses the application factory pattern in app/__init__.py:26:
app/__init__.py
def create_app(config_name=None):
    app = Flask(__name__)

    # Load configuration
    from app.config import get_config
    app.config.from_object(get_config(config_name))

    # Set custom JSON encoder
    app.json_encoder = CustomJSONEncoder

    # Initialize extensions
    db.init_app(app)
    jwt.init_app(app)
    CORS(app)

    # Register blueprints
    from app.routes.auth import auth_bp
    from app.routes.products import products_bp
    from app.routes.orders import orders_bp
    from app.routes.payments import payments_bp

    app.register_blueprint(auth_bp, url_prefix="/api/auth")
    app.register_blueprint(products_bp, url_prefix="/api/products")
    app.register_blueprint(orders_bp, url_prefix="/api/orders")
    app.register_blueprint(payments_bp, url_prefix="/api/payments")

    # Create tables
    with app.app_context():
        from app.models import user, product, order
        db.create_all()

    return app

​
Dependencies

From requirements.txt:
requirements.txt
Flask>=2.3.0
Flask-SQLAlchemy==3.1.1
Flask-JWT-Extended==4.6.0
Flask-Cors==4.0.0
psycopg2-binary==2.9.9
bcrypt==4.1.2
marshmallow==3.20.1
python-dotenv==1.0.0
pytest==7.4.3
pytest-flask==1.3.0
gunicorn==21.2.0
redis==5.0.1

Build docs developers (and LLMs) love

Get started for freeTalk to us