Complete Authentication Example

Building a secure authentication system from scratch can take days. Between password hashing, JWT tokens, user management, and protecting routes, there's a lot to get right. This example shows you how to build a production-ready auth system with Ravyn and Mongoz in minutes, not days.

Let's build a complete user authentication system with registration, login, and protected routes using MongoDB.

What You'll Learn

• Creating user documents with Mongoz
• Building registration and login APIs
• Implementing JWT authentication
• Protecting routes with middleware
• Refreshing JWT tokens
• Testing the complete flow

What We're Building

A complete authentication system with:

• User Registration - Create new accounts
• User Login - Authenticate and get JWT token
• Protected Routes - Access user-specific data
• Token Refresh - Keep users logged in
• Password Security - Automatic hashing with bcrypt

---

Project Structure

We'll organize our code like this:

myapp/
├── accounts/
│   ├── documents.py     # User document
│   └── controllers.py   # API endpoints
├── settings.py          # App configuration
└── app.py              # Main application

Assumptions: - Documents are in accounts/documents.py - Controllers/APIs are in accounts/controllers.py - Main app is in app.py - JWTConfig is in your global settings

Tip

For local development, use Docker for MongoDB:

docker run -d -p 27017:27017 --name mongodb mongo:latest

Let's go!

---

Step 1: Create User Document

First, define a User document using Ravyn's built-in authentication document:

accounts/documents.py

from datetime import datetime
from enum import Enum

import mongoz
from edgy import Registry

from ravyn.contrib.auth.edgy.base_user import User as BaseUser

database = "mongodb://localhost:27017"
registry = Registry(database)


class UserType(str, Enum):
    ADMIN = "admin"
    USER = "user"
    OTHER = "other"


class Role(mongoz.EmbeddedDocument):
    name: str = mongoz.String(max_length=255, default=UserType.USER)


class User(BaseUser):
    """
    Inherits from the BaseUser all the fields and adds extra unique ones.
    """

    date_of_birth: datetime = mongoz.Date()
    is_verified: bool = mongoz.Boolean(default=False)
    role: Role = mongoz.Embed(Role)

    class Meta:
        registry = registry
        database = "my_db"

    def __str__(self):
        return f"{self.email} - {self.role.name}"

This gives you a complete User document with password hashing, validation, and all the fields you need.

---

Step 2: Create User API

Build an API endpoint to register new users:

accounts/controllers.py

from accounts.models import User
from pydantic import BaseModel

from ravyn import post


class UserIn(BaseModel):
    first_name: str
    last_name: str
    email: str
    password: str
    username: str


@post(tags=["user"])
async def create_user(data: UserIn) -> None:
    """
    Creates a user in the system and returns the default 201
    status code.
    """
    await User.create_user(
        first_name=data.first_name,
        last_name=data.last_name,
        email=data.email,
        password=data.password,
        username=data.username,
    )

What this does: - Accepts user registration data - Validates email and password - Creates user with hashed password - Returns success response

Note

This example doesn't handle edge cases like duplicate emails. In production, add proper error handling for unique constraint violations.

---

Step 3: Login API

Create an endpoint that authenticates users and returns a JWT token:

accounts/controllers.py

from datetime import datetime, timedelta

from accounts.models import User
from mongoz.exceptions import DocumentNotFound
from pydantic import BaseModel

from ravyn import JSONResponse, post, status
from ravyn.conf import settings
from ravyn.security.jwt.token import Token


class LoginIn(BaseModel):
    email: str
    password: str


class BackendAuthentication(BaseModel):
    """
    Utility class that helps with the authentication process.
    """

    email: str
    password: str

    async def authenticate(self) -> str:
        """Authenticates a user and returns a JWT string"""
        try:
            user: User = await User.objects.get(email=self.email)
        except DocumentNotFound:
            # Run the default password hasher once to reduce the timing
            # difference between an existing and a nonexistent user.
            await User().set_password(self.password)
        else:
            is_password_valid = await user.check_password(self.password)
            if is_password_valid and self.is_user_able_to_authenticate(user):
                # The lifetime of a token should be short, let us make 5 minutes.
                # You can use also the access_token_lifetime from the JWT config directly
                time = datetime.now() + settings.jwt_config.access_token_lifetime
                return self.generate_user_token(user, time=time)

    def is_user_able_to_authenticate(self, user):
        """
        Reject users with is_active=False. Custom user models that don't have
        that attribute are allowed.
        """
        return getattr(user, "is_active", True)

    def generate_user_token(self, user: User, time=None):
        """
        Generates the JWT token for the authenticated user.
        """
        if not time:
            later = datetime.now() + timedelta(minutes=20)
        else:
            later = time

        token = Token(sub=user.id, exp=later)
        return token.encode(
            key=settings.jwt_config.signing_key, algorithm=settings.jwt_config.algorithm
        )


@post(status_code=status.HTTP_200_OK, tags=["auth"])
async def login(data: LoginIn) -> JSONResponse:
    """
    Login a user and returns a JWT token, else raises ValueError
    """
    auth = BackendAuthentication(email=data.email, password=data.password)
    token = await auth.authenticate()
    return JSONResponse({settings.jwt_config.access_token_name: token})

What's happening here:

1. Receive credentials - Email and password from the request
2. Validate data - BackendAuthentication uses Pydantic for validation
3. Authenticate user - Checks password and generates JWT
4. Return token - Client uses this for subsequent requests

The BackendAuthentication class handles all the heavy lifting: - Validates credentials - Checks password hash - Generates JWT token - Returns token or raises error

Warning

Make sure your JWTConfig is configured in your global settings as assumed at the top of this document.

---

Step 4: Protected Home API

Create an endpoint that requires authentication:

accounts/controllers.py

from ravyn import JSONResponse, Request, get


@get(tags=["home"])
async def home(request: Request) -> JSONResponse:
    """
    Ravyn request has a `user` property that also
    comes from its origins (Lilya).

    When building an authentication middleware, it
    is recommended to inherit from the `BaseAuthMiddleware`.

    See more info here: https://ravyn.dymmond.com/middleware/middleware/?h=baseauthmiddleware#baseauthmiddleware
    """
    return JSONResponse({"message": f"hello {request.user.email}"})

Simple and clean: - Middleware validates JWT token - User is automatically injected into request.user - Return user-specific data

---

Step 5: Assemble the Application

Put it all together in your main application:

app.py

#!/usr/bin/env python
"""
Generated by 'ravyn createproject'
"""

import os
import sys
from pathlib import Path

from ravyn import Ravyn, Gateway, Include
from ravyn.conf import settings
from ravyn.contrib.auth.mongoz.middleware import JWTAuthMiddleware, JWTAuthBackend
from lilya.middleware import DefineMiddleware as LilyaMiddleware


def build_path():
    """
    Builds the path of the project and project root.
    """
    Path(__file__).resolve().parent.parent
    SITE_ROOT = os.path.dirname(os.path.realpath(__file__))

    if SITE_ROOT not in sys.path:
        sys.path.append(SITE_ROOT)
        sys.path.append(os.path.join(SITE_ROOT, "apps"))


def get_application():
    """
    This is optional. The function is only used for organisation purposes.
    """
    build_path()
    from accounts.models import User
    from accounts.views import create_user, home, login

    app = Ravyn(
        routes=[
            Gateway("/login", handler=login),
            Gateway("/create", handler=create_user),
            Include(
                routes=[Gateway(handler=home)],
                middleware=[
                    LilyaMiddleware(
                        JWTAuthMiddleware,
                        backend=JWTAuthBackend(
                            config=settings.jwt_config,
                            user_model=User,
                        ),
                    ),
                ],
            ),
        ],
    )
    return app


app = get_application()

Notice the middleware placement:

The JWTAuthMiddleware is inside the Include for /, not in the main Ravyn instance. This is intentional!

• Public routes (/create, /login) - No authentication needed
• Protected routes (/) - Authentication required

Each Include can have its own middleware, giving you fine-grained control over which routes require authentication.

---

Step 6: Refreshing Tokens

JWT tokens expire for security. Implement token refresh to keep users logged in without re-entering credentials.

Ravyn provides a complete guide for implementing refresh tokens:

See refresh token implementation →

Key concepts: - Issue both access and refresh tokens - Access tokens expire quickly (15 minutes) - Refresh tokens last longer (7 days) - Use refresh token to get new access token

---

Testing the Flow

Let's test the complete authentication flow using httpx:

Complete Test Script

import httpx

# The password is automatically encrypted when using the
# User model provided by Ravyn
user_data = {
    "first_name": "John",
    "last_name": "Doe",
    "email": "john@doe.com",
    "username": "john.doe",
    "password": "johnspassword1234@!",
}

# Create a user
# This returns a 201
async with httpx.AsyncClient() as client:
    client.post("/create", json=user_data)

# Login the user
# Returns the response with the JWT token
user_login = {"email": user_data["email"], "password": user_data["password"]}

async with httpx.AsyncClient() as client:
    response = client.post("/login", json=user_login)

# Access the home '/' endpoint
# The default header for the JWTConfig used is `Authorization``
# The default auth_header_types of the JWTConfig is ["Bearer"]
access_token = response.json()["access_token"]

async with httpx.AsyncClient() as client:
    response = client.get("/", headers={"Authorization": f"Bearer {access_token}"})

print(response.json()["message"])
# hello john@doe.com

Step-by-Step

1. Create a user:

response = client.post("/create", json={
    "email": "user@example.com",
    "password": "securepass123"
})

2. Login and get token:

response = client.post("/login", json={
    "email": "user@example.com",
    "password": "securepass123"
})
token = response.json()["token"]

3. Access protected route:

response = client.get("/", headers={
    "Authorization": f"Bearer {token}"
})

The Authorization header:

Notice the Authorization header format: Bearer {token}. This is the default expected by JWTConfig. The header name (Authorization) is configurable in JWTConfig.

---

Common Pitfalls & Fixes

Pitfall 1: Middleware on Wrong Routes

Problem: Auth middleware on login/register routes.

# Wrong - can't login if auth is required!
app = Ravyn(
    middleware=[JWTAuthMiddleware(...)],  # Applies to ALL routes
    routes=[...]
)

Solution: Apply middleware selectively:

# Correct - auth only where needed
app = Ravyn(routes=[
    Gateway(handler=create_user),  # No auth
    Gateway(handler=login),        # No auth
    Include(
        routes=[Gateway(handler=home)],
        middleware=[JWTAuthMiddleware(...)]  # Auth required
    )
])

Pitfall 2: Missing Token in Requests

Problem: Forgetting to include JWT token.

// Wrong
fetch('/api/profile')

Solution: Always include Authorization header:

// Correct
fetch('/api/profile', {
    headers: {
        'Authorization': `Bearer ${token}`
    }
})

Pitfall 3: MongoDB Connection Not Closed

Problem: Not properly closing MongoDB connections.

Solution: Use context managers or lifespan events:

from contextlib import asynccontextmanager

@asynccontextmanager
async def lifespan(app):
    # Startup
    await client.connect()
    yield
    # Shutdown
    await client.disconnect()

app = Ravyn(lifespan=lifespan)

---

Best Practices

1. Use Environment Variables

# settings.py
import os

class AppSettings(RavynSettings):
    jwt_secret: str = os.getenv("JWT_SECRET")
    mongodb_url: str = os.getenv("MONGODB_URL", "mongodb://localhost:27017")

2. Validate Input Data

from pydantic import BaseModel, EmailStr

class UserCreate(BaseModel):
    email: EmailStr
    password: str

    @validator('password')
    def password_strength(cls, v):
        if len(v) < 8:
            raise ValueError('Password too short')
        return v

3. Handle Errors Gracefully

@post("/login")
async def login(data: dict) -> JSONResponse:
    try:
        auth = BackendAuthentication(**data)
        token = await auth.authenticate()
        return JSONResponse({"token": token})
    except AuthenticationError:
        return JSONResponse(
            {"error": "Invalid credentials"},
            status_code=401
        )

4. Use Indexes for Performance

class User(AbstractUser):
    class Meta:
        registry = registry
        database = "myapp"
        indexes = [
            mongoz.Index("email", unique=True),
            mongoz.Index("username", unique=True)
        ]

5. Use HTTPS in Production

Always use HTTPS to protect JWT tokens in transit.

---

MongoDB Advantages

No Migrations

Unlike SQL, you can modify your schema without migrations:

# Add fields anytime
class User(AbstractUser):
    phone = fields.String()  # Just add it!
    preferences = fields.Object()  # No migration needed

Flexible Data

Store varying user data:

# Different users can have different fields
user1 = User(email="user1@example.com", bio="Short")
user2 = User(email="user2@example.com", social={...})

---

Conclusion

You've built a complete, production-ready authentication system with:

✅ User registration with password hashing ✅ Login with JWT token generation
✅ Protected routes with middleware ✅ Clean, maintainable code structure ✅ MongoDB flexibility

This is just the beginning. You can extend this with: - Email verification - Password reset - Role-based permissions - OAuth integration - Two-factor authentication

---

Learn More

• User Documents - Deep dive into user documents
• JWT Middleware - Middleware configuration
• JWTConfig - JWT configuration options
• Refresh Tokens - Token refresh implementation

---

Next Steps

• Edgy Example - SQL authentication
• Permissions - Add role-based access
• Testing - Test your authentication