# Python for PHP Developers, Part 5: Refresh Tokens in FastAPI

*This is Part 5 of a series. Catch up on* [*Part 1: A Practical Crash Course*](https://himanitayal.hashnode.dev/python-for-php-developers-a-practical-crash-course)*,* [*Part 2: FastAPI with a Real Database*](https://himanitayal.hashnode.dev/python-for-php-part-2-fastapi-with-database)*,* [*Part 3: JWT Authentication*](PART-3-LINK)*, and* [*Part 4: Testing FastAPI with pytest*](https://himanitayal.hashnode.dev/python-for-php-developers-part-4-testing-fastapi-with-pytest)*.*

In Part 3, we issued a 30-minute access token. That's good for security, but users get logged out every half hour. I also mentioned that JWTs aren't revocable by default. This time, we fix both problems with refresh tokens.

## 1\. Why two tokens?

|  | Access token | Refresh token |
| --- | --- | --- |
| Lifetime | Short (30 minutes) | Long (7 days) |
| Sent with | Every API request | Only to `/refresh` |
| Stored on server? | No | Yes (so it can be revoked) |
| If stolen | Useful for minutes | Dangerous, so we guard it |

The access token stays stateless and fast. The refresh token is the one we keep track of, so we can cancel it.

## 2\. Update the token helpers

We make two changes to `auth.py` from Part 3: a `type` claim, so a refresh token can never be used as an access token, and a new function that creates refresh tokens with a unique ID (`jti`).

```python
# auth.py (updated)
import uuid
from datetime import datetime, timedelta, timezone
from jose import jwt, JWTError

SECRET_KEY = "change-this-in-production"   # load from env, never hardcode
ALGORITHM = "HS256"
ACCESS_TOKEN_EXPIRE_MINUTES = 30
REFRESH_TOKEN_EXPIRE_DAYS = 7

def create_access_token(data: dict, expires_delta: timedelta | None = None) -> str:
    to_encode = data.copy()
    expire = datetime.now(timezone.utc) + (
        expires_delta or timedelta(minutes=ACCESS_TOKEN_EXPIRE_MINUTES)
    )
    to_encode.update({"exp": expire, "type": "access"})
    return jwt.encode(to_encode, SECRET_KEY, algorithm=ALGORITHM)

def create_refresh_token(username: str):
    jti = str(uuid.uuid4())
    expire = datetime.now(timezone.utc) + timedelta(days=REFRESH_TOKEN_EXPIRE_DAYS)
    payload = {"sub": username, "jti": jti, "type": "refresh", "exp": expire}
    token = jwt.encode(payload, SECRET_KEY, algorithm=ALGORITHM)
    return token, jti, expire

def decode_token(token: str) -> dict:
    # Verifies signature and expiry. The "type" claim is checked by the caller.
    try:
        return jwt.decode(token, SECRET_KEY, algorithms=[ALGORITHM])
    except JWTError:
        raise ValueError("Invalid or expired token")
```

I renamed `decode_access_token` to `decode_token`, since it now decodes both kinds.

## 3\. Store refresh tokens server-side

```python
# models.py (addition)
from sqlalchemy import Column, Integer, String, Boolean, DateTime
from database import Base

class RefreshToken(Base):
    __tablename__ = "refresh_tokens"

    id = Column(Integer, primary_key=True, index=True)
    jti = Column(String, unique=True, index=True)
    username = Column(String, index=True)
    expires_at = Column(DateTime)
    revoked = Column(Boolean, default=False)
```

We store only the `jti`, not the token itself. A leaked database row can't be turned into a working token, because the signature lives in the token, not in the row.

```python
# schemas.py (addition)
class RefreshRequest(BaseModel):
    refresh_token: str
```

Send the refresh token in the request body, not as a query parameter. Query strings end up in server logs.

## 4\. Login returns both tokens

```python
# main.py (updated login)
from auth import create_access_token, create_refresh_token, decode_token

@app.post("/login")
def login(form_data: OAuth2PasswordRequestForm = Depends(), db: Session = Depends(get_db)):
    user = db.query(models.User).filter(models.User.username == form_data.username).first()
    if not user or not verify_password(form_data.password, user.hashed_password):
        raise HTTPException(status_code=401, detail="Incorrect username or password")

    access_token = create_access_token({"sub": user.username})
    refresh_token, jti, expires_at = create_refresh_token(user.username)
    db.add(models.RefreshToken(jti=jti, username=user.username, expires_at=expires_at))
    db.commit()

    return {
        "access_token": access_token,
        "refresh_token": refresh_token,
        "token_type": "bearer",
    }
```

## 5\. The `/refresh` endpoint: rotation and reuse detection

Every time a refresh token is used, we revoke it and issue a brand new pair. This is called **rotation**. If an already-used token ever shows up again, something is wrong (it may have been stolen), so we revoke all of that user's tokens.

```python
@app.post("/refresh")
def refresh(body: schemas.RefreshRequest, db: Session = Depends(get_db)):
    try:
        payload = decode_token(body.refresh_token)
    except ValueError:
        raise HTTPException(status_code=401, detail="Invalid refresh token")

    if payload.get("type") != "refresh":
        raise HTTPException(status_code=401, detail="Wrong token type")

    stored = (
        db.query(models.RefreshToken)
        .filter(models.RefreshToken.jti == payload["jti"])
        .first()
    )
    if not stored:
        raise HTTPException(status_code=401, detail="Unknown refresh token")

    if stored.revoked:
        # A used token came back: possible theft. Revoke everything for this user.
        db.query(models.RefreshToken).filter(
            models.RefreshToken.username == stored.username
        ).update({"revoked": True})
        db.commit()
        raise HTTPException(status_code=401, detail="Refresh token reuse detected")

    # Rotate: revoke the old token, issue a new pair
    stored.revoked = True
    new_refresh, new_jti, expires_at = create_refresh_token(payload["sub"])
    db.add(models.RefreshToken(jti=new_jti, username=payload["sub"], expires_at=expires_at))
    db.commit()

    return {
        "access_token": create_access_token({"sub": payload["sub"]}),
        "refresh_token": new_refresh,
        "token_type": "bearer",
    }
```

## 6\. Logout that actually works

```python
@app.post("/logout")
def logout(body: schemas.RefreshRequest, db: Session = Depends(get_db)):
    try:
        payload = decode_token(body.refresh_token)
    except ValueError:
        return {"message": "Logged out"}

    stored = (
        db.query(models.RefreshToken)
        .filter(models.RefreshToken.jti == payload.get("jti"))
        .first()
    )
    if stored:
        stored.revoked = True
        db.commit()
    return {"message": "Logged out"}
```

After logout, the refresh token is dead, so the session can't be renewed.

## 7\. Lock down the access check

Update `get_current_user` from Part 3 so it rejects refresh tokens:

```python
def get_current_user(token: str = Depends(oauth2_scheme)):
    try:
        payload = decode_token(token)
    except ValueError:
        raise HTTPException(status_code=401, detail="Could not validate credentials")

    if payload.get("type") != "access":
        raise HTTPException(status_code=401, detail="Wrong token type")
    return payload["sub"]
```

## 8\. Test it

This builds on the `client` fixture from Part 4.

```python
# tests/test_refresh.py
from datetime import timedelta
from auth import create_access_token

def login_tokens(client):
    client.post("/register", params={"username": "himani", "password": "secret123"})
    response = client.post(
        "/login", data={"username": "himani", "password": "secret123"}
    )
    return response.json()

def test_refresh_returns_new_tokens(client):
    tokens = login_tokens(client)
    response = client.post("/refresh", json={"refresh_token": tokens["refresh_token"]})
    assert response.status_code == 200
    assert response.json()["refresh_token"] != tokens["refresh_token"]

def test_refresh_token_cannot_be_reused(client):
    tokens = login_tokens(client)
    client.post("/refresh", json={"refresh_token": tokens["refresh_token"]})
    second = client.post("/refresh", json={"refresh_token": tokens["refresh_token"]})
    assert second.status_code == 401

def test_refresh_token_rejected_as_access_token(client):
    tokens = login_tokens(client)
    response = client.post(
        "/items",
        json={"name": "Pen", "price": 1.5},
        headers={"Authorization": f"Bearer {tokens['refresh_token']}"},
    )
    assert response.status_code == 401

def test_expired_access_token_rejected(client):
    expired = create_access_token({"sub": "himani"}, timedelta(minutes=-1))
    response = client.post(
        "/items",
        json={"name": "Pen", "price": 1.5},
        headers={"Authorization": f"Bearer {expired}"},
    )
    assert response.status_code == 401
```

The `expires_delta` parameter we added earlier makes the expiry test easy: we create a token that expired a minute ago instead of waiting 30 minutes.

## 9\. Gotchas for PHP developers

- **Access tokens still can't be revoked instantly.** Refresh tokens fix the long-lived session, but a stolen access token stays valid until it expires. That's why we keep it short.
- **Always check the** `type` **claim.** Without it, a long-lived refresh token could be used as an access token and bypass the whole design.
- **Rotation plus reuse detection is the safety net.** A stolen refresh token only works once, and the legitimate user's next refresh exposes the theft.
- **Where the client stores tokens matters.** For browser apps, avoid `localStorage` for refresh tokens, since any script injected into the page can read it. An `HttpOnly` cookie is the usual choice.
- **Clean up old rows.** Expired and revoked tokens pile up in the table. A scheduled job (a cron task, if you're coming from PHP) that deletes expired rows keeps it small.

## What's next

The API now has login, refresh, logout, and tests for all of it. Natural next steps are role-based permissions, rate limiting on `/login`, and running the tests automatically in CI.

*How does your team handle token refresh? Let me know in the comments.*