python-modal

name

python-modal

description

Modern Python patterns for Modal.com serverless platform. PROACTIVELY activate for: (1) Modal function deployment, (2) Type-safe Modal with Pydantic, (3) Async patterns in Modal, (4) GPU workloads (ML inference, training), (5) FastAPI web endpoints on Modal, (6) Scheduled tasks and cron jobs, (7) Modal Volumes and storage, (8) Testing Modal functions with pytest, (9) Modal classes and lifecycle methods, (10) Parallel processing with map/starmap. Provides: Type hints patterns, Pydantic integration, async/await patterns, pytest testing, FastAPI integration, scheduled tasks, Volume usage, cost optimization, and production-ready examples following Python 3.11+ best practices.

Quick Reference

Pattern	Decorator	Use Case
Simple function	@app.function()	Stateless compute
Class with state	@app.cls()	ML models, DB connections
Web endpoint	@modal.asgi_app()	FastAPI/Flask APIs
Scheduled	@app.function(schedule=...)	Cron jobs
Parallel	.map(), .starmap()	Batch processing

When to Use This Skill

Use for Python on Modal:

• Type-safe serverless functions with Pydantic
• Async/await patterns for concurrent operations
• GPU workloads (ML inference, training)
• FastAPI web endpoints
• Scheduled tasks and automation
• Testing Modal functions locally

---

Modern Python on Modal.com (2025)

Complete guide to building type-safe, production-ready Python applications on Modal's serverless platform.

Type-Safe Modal Functions

Basic Function with Type Hints

import modal

app = modal.App("typed-example")

image = modal.Image.debian_slim(python_version="3.11").pip_install("pydantic")

@app.function(image=image)
def process_data(
    items: list[str],
    multiplier: int = 1,
) -> dict[str, int]:
    """Process data with full type hints."""
    return {item: len(item) * multiplier for item in items}


@app.local_entrypoint()
def main():
    result: dict[str, int] = process_data.remote(["hello", "world"], multiplier=2)
    print(result)  # {'hello': 10, 'world': 10}

Pydantic Models for Validation

import modal
from pydantic import BaseModel, Field, field_validator
from datetime import datetime
from typing import Literal

app = modal.App("pydantic-example")

image = modal.Image.debian_slim(python_version="3.11").pip_install("pydantic>=2.0")


# === Input/Output Models ===

class ProcessingConfig(BaseModel):
    """Configuration with validation."""
    model_name: str = Field(..., min_length=1, max_length=100)
    batch_size: int = Field(32, ge=1, le=256)
    temperature: float = Field(0.7, ge=0.0, le=2.0)
    mode: Literal["fast", "balanced", "quality"] = "balanced"

    @field_validator("model_name")
    @classmethod
    def validate_model_name(cls, v: str) -> str:
        allowed = ["gpt2", "llama-7b", "mistral-7b"]
        if v not in allowed:
            raise ValueError(f"Model must be one of ")
         v


 ():
    
    text:  = Field(..., min_length=, max_length=)
    config: ProcessingConfig
    user_id:  = Field(..., pattern=)


 ():
    
    result: 
    tokens_used: 
    latency_ms: 
    model_name: 
    timestamp: datetime



 () -> ProcessingResponse:
    
     time

    start = time.time()

    
    result = 
    tokens = (request.text.split())

    latency = (time.time() - start) * 

     ProcessingResponse(
        result=result,
        tokens_used=tokens,
        latency_ms=(latency, ),
        model_name=request.config.model_name,
        timestamp=datetime.utcnow(),
    )



 ():
    
    request = ProcessingRequest(
        text=,
        config=ProcessingConfig(model_name=, batch_size=),
        user_id=,
    )

    response = process_with_validation.remote(request)
    ()
    ()

---

Async Patterns

Async Modal Functions

import modal
import asyncio
from typing import Coroutine, Any

app = modal.App("async-example")

image = modal.Image.debian_slim(python_version="3.11").pip_install("httpx", "aiofiles")


@app.function(image=image)
async def fetch_url(url: str) -> dict[str, Any]:
    """Async function for HTTP requests."""
    import httpx

    async with httpx.AsyncClient() as client:
        response = await client.get(url, timeout=30)
        return {
            "url": url,
            "status": response.status_code,
            "length": len(response.content),
        }


@app.function(image=image)
async def fetch_multiple(urls: list[str]) -> list[dict[str, Any]]:
    """Fetch multiple URLs concurrently."""
    import httpx

    async with httpx.AsyncClient() as client:
        tasks = [client.get(url, timeout=30)  url  urls]
        responses =  asyncio.gather(*tasks, return_exceptions=)

        results = []
         url, response  (urls, responses):
             (response, Exception):
                results.append({: url, : (response)})
            :
                results.append({
                    : url,
                    : response.status_code,
                    : (response.content),
                })

         results



  () -> []:
    
     aiofiles

      () -> :
          aiofiles.(path, )  f:
              f.read()

    tasks = [read_file(path)  path  file_paths]
    contents =  asyncio.gather(*tasks)
     contents



 ():
    urls = [
        ,
        ,
        ,
    ]

    results = fetch_multiple.remote(urls)
     result  results:
        ()

Async with Modal Classes

import modal

app = modal.App("async-class")

image = modal.Image.debian_slim(python_version="3.11").pip_install("httpx", "redis")


@app.cls(image=image)
class AsyncService:
    """Service with async methods and connection pooling."""

    def __init__(self):
        self._http_client: httpx.AsyncClient | None = None

    @modal.enter()
    async def setup(self):
        """Async initialization - runs once per container."""
        import httpx
        self._http_client = httpx.AsyncClient(timeout=30)
        print("HTTP client initialized")

    @modal.exit()
    async def cleanup(self):
        """Async cleanup - runs on container shutdown."""
        if self._http_client:
            await self._http_client.aclose()
        print("HTTP client closed")

    @modal.method()
    async def fetch(self, url: str) -> dict:
        """Use persistent HTTP client."""
        response =  ._http_client.get(url)
         {: url, : response.status_code}


      () -> []:
        
         asyncio

        tasks = [._http_client.get(url)  url  urls]
        responses =  asyncio.gather(*tasks, return_exceptions=)

         [
            {: url, : r.status_code}
              (r, Exception)
             {: url, : (r)}
             url, r  (urls, responses)
        ]

---

Modal Classes with Lifecycle

ML Model Server

import modal
from pydantic import BaseModel, Field

app = modal.App("ml-server")

image = (
    modal.Image.debian_slim(python_version="3.11")
    .pip_install("torch", "transformers", "pydantic")
)

models_volume = modal.Volume.from_name("models-cache", create_if_missing=True)


class InferenceRequest(BaseModel):
    text: str = Field(..., min_length=1, max_length=5000)
    max_tokens: int = Field(512, ge=1, le=2048)
    temperature: float = Field(0.7, ge=0.0, le=2.0)


class InferenceResponse(BaseModel):
    generated_text: str
    tokens_generated: int
    model_name: str


@app.cls(
    image=image,
    gpu="A100",
    volumes={"/models": models_volume},
    min_containers=1,      # Keep warm
    max_containers=10,     # Scale limit
    container_idle_timeout=300,
)
class ModelServer:
    """Type-safe ML model server with lifecycle management."""

    model_name: str = "gpt2"

    def __init__(self, model_name:  = ):
        .model_name = model_name
        ._model = 
        ._tokenizer = 


     () -> :
        
         transformers  AutoModelForCausalLM, AutoTokenizer
         torch

        ()

        ._tokenizer = AutoTokenizer.from_pretrained(
            .model_name,
            cache_dir=,
        )
        ._model = AutoModelForCausalLM.from_pretrained(
            .model_name,
            torch_dtype=torch.float16,
            cache_dir=,
        ).to()

        ()


     () -> InferenceResponse:
        
         torch

        inputs = ._tokenizer(request.text, return_tensors=).to()

         torch.no_grad():
            outputs = ._model.generate(
                **inputs,
                max_new_tokens=request.max_tokens,
                temperature=request.temperature,
                do_sample=,
            )

        generated = ._tokenizer.decode(outputs[], skip_special_tokens=)
        tokens_generated = (outputs[]) - (inputs.input_ids[])

         InferenceResponse(
            generated_text=generated,
            tokens_generated=tokens_generated,
            model_name=.model_name,
        )


     () -> [InferenceResponse]:
        
         [.generate(req)  req  requests]



 ():
    server = ModelServer()

    request = InferenceRequest(
        text=,
        max_tokens=,
        temperature=,
    )

    response = server.generate.remote(request)
    ()
    ()

---

FastAPI Integration

Type-Safe Web API

import modal
from pydantic import BaseModel, Field
from datetime import datetime
from typing import Annotated

app = modal.App("fastapi-example")

image = (
    modal.Image.debian_slim(python_version="3.11")
    .pip_install("fastapi", "pydantic>=2.0")
)


# === Pydantic Models ===

class ItemCreate(BaseModel):
    name: str = Field(..., min_length=1, max_length=100)
    description: str = Field("", max_length=1000)
    price: float = Field(..., gt=0)
    tags: list[str] = []


class Item(ItemCreate):
    id: str
    created_at: datetime


class ItemList(BaseModel):
    items: list[Item]
    total: int


# === In-memory storage (use database in production) ===
items_db: dict[str, Item] = {}


@app.function(image=image)
@modal.concurrent(max_inputs=100, target_inputs=50)
@modal.asgi_app()
def api():
    from fastapi  FastAPI, HTTPException, Query, Path, Depends
     fastapi.security  HTTPBearer, HTTPAuthorizationCredentials
     uuid

    web_app = FastAPI(
        title=,
        version=,
        description=,
    )

    security = HTTPBearer()

     () -> :
         creds.credentials != :
             HTTPException(, )
         creds.credentials

    


     () -> Item:
        
        item_id = (uuid.uuid4())
        new_item = Item(
            =item_id,
            created_at=datetime.utcnow(),
            **item.model_dump(),
        )
        items_db[item_id] = new_item
         new_item


     () -> ItemList:
        
        all_items = (items_db.values())
         ItemList(
            items=all_items[skip : skip + limit],
            total=(all_items),
        )


     () -> Item:
        
         item_id   items_db:
             HTTPException(, )
         items_db[item_id]


     () -> :
        
         item_id   items_db:
             HTTPException(, )
         items_db[item_id]


     () -> [, ]:
         {: , : datetime.utcnow().isoformat()}

     web_app

---

Scheduled Tasks

Cron Jobs with Type Safety

import modal
from pydantic import BaseModel
from datetime import datetime, timezone
import os

app = modal.App("scheduled-tasks")

image = modal.Image.debian_slim(python_version="3.11").pip_install(
    "httpx",
    "pydantic",
)


class ETLResult(BaseModel):
    """Result of ETL job."""
    records_processed: int
    errors: int
    duration_seconds: float
    timestamp: datetime


@app.function(
    image=image,
    schedule=modal.Cron("0 6 * * *", timezone="America/New_York"),  # 6 AM ET daily
    secrets=[modal.Secret.from_name("api-credentials")],
    timeout=3600,
)
def daily_etl_job() -> ETLResult:
    """Daily ETL job that runs at 6 AM."""
    import httpx
    import time

    start = time.time()
    api_key = os.environ["API_KEY"]

    records = 0
    errors = 0

    try:
        with httpx.Client() as client:
            response = client.get(
                "https://api.example.com/data",
                headers={"Authorization": f"Bearer {api_key}"},
                timeout=60,
            )
            response.raise_for_status()
            data = response.json()
            records = len(data.get(, []))
     httpx.HTTPError  e:
        errors += 
        ()

    duration = time.time() - start

     ETLResult(
        records_processed=records,
        errors=errors,
        duration_seconds=(duration, ),
        timestamp=datetime.now(timezone.utc),
    )



 () -> [, ]:
    
     httpx

    endpoints = [
        ,
        ,
    ]

    results = {}
     httpx.Client()  client:
         url  endpoints:
            :
                response = client.get(url, timeout=)
                results[url] =   response.status_code ==   
             httpx.HTTPError:
                results[url] = 

     results



 () -> [, ]:
    
     random  

     {
        : random.uniform(, ),
        : random.uniform(, ),
        : random.randint(, ),
        : datetime.now(timezone.utc).isoformat(),
    }

---

Parallel Processing

Map and Starmap with Types

import modal
from pydantic import BaseModel
from typing import Iterator

app = modal.App("parallel-processing")

image = modal.Image.debian_slim(python_version="3.11").pip_install("pydantic", "httpx")


class ProcessingInput(BaseModel):
    id: str
    data: str


class ProcessingOutput(BaseModel):
    id: str
    result: str
    success: bool


@app.function(image=image, timeout=300)
def process_single(input_data: ProcessingInput) -> ProcessingOutput:
    """Process a single item."""
    try:
        # Your processing logic
        result = f"Processed: {input_data.data.upper()}"
        return ProcessingOutput(id=input_data.id, result=result, success=True)
    except Exception as e:
        return ProcessingOutput(id=input_data.id, result=str(e), success=False)


@app.function(image=image)
def process_batch() -> [ProcessingOutput]:
    
    
    results = (process_single.(inputs))
     results



 () -> :
    
     



 () -> []:
    
    
    args = ((data_list, multipliers, prefixes))
    results = (process_with_args.starmap(args))
     results



 () -> Iterator[ProcessingOutput]:
    
     () -> Iterator[ProcessingInput]:
         i  ():
             ProcessingInput(=(i), data=)

    
     result  process_single.(input_generator()):
         result



 ():
    
    inputs = [
        ProcessingInput(=, data=),
        ProcessingInput(=, data=),
        ProcessingInput(=, data=),
    ]

    results = process_batch.remote(inputs)
     result  results:
        ()

    
    data = [, , ]
    multipliers = [, , ]
    prefixes = [, , ]

    starmap_results = process_batch_with_args.remote(data, multipliers, prefixes)
    (starmap_results)  

---

Modal Volumes

Type-Safe Volume Operations

import modal
from pydantic import BaseModel
from pathlib import Path
from datetime import datetime
import json

app = modal.App("volume-example")

image = modal.Image.debian_slim(python_version="3.11").pip_install("pydantic")

# Create persistent volume
data_volume = modal.Volume.from_name("app-data", create_if_missing=True)


class DataRecord(BaseModel):
    """Data record with validation."""
    id: str
    content: str
    created_at: datetime
    metadata: dict[str, str] = {}


class StorageResult(BaseModel):
    """Result of storage operation."""
    success: bool
    path: str
    size_bytes: int
    message: str


@app.function(
    image=image,
    volumes={"/data": data_volume},
)
def save_record(record: DataRecord) -> StorageResult:
    """Save record to volume."""
    file_path = Path(f"/data/records/{record.id}.json")
    file_path.parent.mkdir(parents=True, exist_ok=True)

    # Write JSON
    content = record.model_dump_json(indent=)
    file_path.write_text(content)

    
    data_volume.commit()

     StorageResult(
        success=,
        path=(file_path),
        size_bytes=(content),
        message=,
    )



 () -> DataRecord | :
    
    file_path = Path()

      file_path.exists():
         

    content = file_path.read_text()
     DataRecord.model_validate_json(content)



 () -> []:
    
    records_dir = Path()

      records_dir.exists():
         []

     [f.stem  f  records_dir.glob()]



 () -> StorageResult:
    
     shutil

    
    volume_input = Path()
    temp_input = Path()
    temp_output = Path()

    shutil.copy(volume_input, temp_input)

    
    content = temp_input.read_bytes()
    processed = content.upper()  
    temp_output.write_bytes(processed)

    
    volume_output = Path()
    shutil.copy(temp_output, volume_output)

    
    data_volume.commit()

     StorageResult(
        success=,
        path=(volume_output),
        size_bytes=(processed),
        message=,
    )



 ():
    
    record = DataRecord(
        =,
        content=,
        created_at=datetime.utcnow(),
        metadata={: },
    )

    result = save_record.remote(record)
    ()

    
    loaded = load_record.remote()
     loaded:
        ()

    
    records = list_records.remote()
    ()

---

Testing Modal Functions

pytest Integration

# tests/test_modal_functions.py
import pytest
from datetime import datetime
from unittest.mock import Mock, patch

# Import your Modal functions
from app import (
    process_data,
    ProcessingRequest,
    ProcessingResponse,
    ProcessingConfig,
)


class TestProcessingModels:
    """Test Pydantic models."""

    def test_valid_config(self):
        config = ProcessingConfig(
            model_name="gpt2",
            batch_size=64,
            temperature=0.8,
        )
        assert config.model_name == "gpt2"
        assert config.batch_size == 64

    def test_invalid_model_name(self):
        with pytest.raises(ValueError, match="Model must be one of"):
            ProcessingConfig(model_name="invalid-model")

    def test_config_defaults(self):
        config = ProcessingConfig(model_name="gpt2")
        assert config.batch_size == 32  # Default
        assert config.temperature == 0.7  # Default
        assert config.mode == "balanced"  # Default

    def test_request_validation(self):
        config = ProcessingConfig(model_name=)
        request = ProcessingRequest(
            text=,
            config=config,
            user_id=,
        )
         request.text == 

     ():
        config = ProcessingConfig(model_name=)
         pytest.raises(ValueError):
            ProcessingRequest(
                text=,
                config=config,
                user_id=,  
            )


 :
    

     ():
        
        
        result = process_data.local([, ], multiplier=)

         (result, )
         result[] ==   
         result[] == 

     ():
        result = process_data.local([], multiplier=)
         result == {}


      ():
        
         app  fetch_url

        
         patch()  mock_client:
            mock_response = Mock()
            mock_response.status_code = 
            mock_response.content = 

            mock_client.return_value.__aenter__.return_value.get.return_value = mock_response

            result =  fetch_url.local()

             result[] == 
             result[] == 


 :
    


     ():
         ProcessingRequest(
            text=,
            config=ProcessingConfig(model_name=),
            user_id=,
        )

     ():
        
         app  process_with_validation

        response = process_with_validation.local(sample_request)

         (response, ProcessingResponse)
         response.model_name == 
         response.tokens_used > 
         response.latency_ms >= 





 ():
    
     patch()  mock:
         mock



 ():
    
     patch(, return_value=):
        

Running Tests

# Run all tests
pytest tests/ -v

# Run with coverage
pytest tests/ --cov=app --cov-report=html

# Run specific test class
pytest tests/test_modal_functions.py::TestProcessingModels -v

# Run async tests
pytest tests/ -v --asyncio-mode=auto

pytest Configuration

# pyproject.toml
[tool.pytest.ini_options]
testpaths = ["tests"]
python_files = ["test_*.py"]
python_functions = ["test_*"]
asyncio_mode = "auto"
addopts = "-v --tb=short"

[tool.coverage.run]
source = ["app"]
omit = ["tests/*"]

---

Best Practices

1. Always Use Type Hints

# Good
def process(items: list[str], count: int = 10) -> dict[str, int]:
    return {item: len(item) for item in items[:count]}

# Bad
def process(items, count=10):
    return {item: len(item) for item in items[:count]}

2. Use Pydantic for Validation

# Good - validated at runtime
class Config(BaseModel):
    batch_size: int = Field(ge=1, le=256)

# Bad - no validation
config = {"batch_size": -1}  # Invalid but accepted

3. Use @modal.enter() for Initialization

# Good - runs once per container
@app.cls(gpu="A100")
class ModelServer:
    @modal.enter()
    def setup(self):
        self.model = load_model()  # Expensive, done once

# Bad - runs every request
@app.function(gpu="A100")
def predict(data):
    model = load_model()  # Loads every time!
    return model.predict(data)

4. Commit Volume Changes

# Good
@app.function(volumes={"/data": vol})
def save(data):
    Path("/data/file.txt").write_text(data)
    vol.commit()  # Persist changes

# Bad - changes lost!
@app.function(volumes={"/data": vol})
def save(data):
    Path("/data/file.txt").write_text(data)
    # Forgot to commit!

5. Use Async for I/O Operations

# Good - concurrent I/O
async def fetch_all(urls: list[str]):
    async with httpx.AsyncClient() as client:
        tasks = [client.get(url) for url in urls]
        return await asyncio.gather(*tasks)

# Bad - sequential I/O
def fetch_all(urls: list[str]):
    results = []
    for url in urls:
        results.append(requests.get(url))  # Blocking!
    return results

---

Common Pitfalls

1. Blocking I/O in Async Functions

# Bad - blocks event loop
async def bad_fetch():
    return requests.get(url)  # Blocking!

# Good - use async client
async def good_fetch():
    async with httpx.AsyncClient() as client:
        return await client.get(url)

2. Not Using @modal.concurrent

# Bad - one request at a time
@app.function()
@modal.asgi_app()
def api():
    return app

# Good - handle multiple requests
@app.function()
@modal.concurrent(max_inputs=100)
@modal.asgi_app()
def api():
    return app

3. Hardcoding Secrets

# Bad
API_KEY = "sk-12345"

# Good
@app.function(secrets=[modal.Secret.from_name("api-keys")])
def process():
    api_key = os.environ["API_KEY"]

4. Forgetting Timeouts

# Bad - default 300s may not be enough
@app.function()
def long_process():
    pass

# Good - explicit timeout
@app.function(timeout=3600)  # 1 hour
def long_process():
    pass

---

Related Skills

• python-type-hints - Comprehensive type hints
• python-asyncio - Async patterns
• python-fastapi - FastAPI best practices
• python-testing - pytest patterns