# FastAPI Marshmallow

***

**1. Serializing and Deserializing Data**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()
    age = fields.Integer()

@app.post("/users")
async def create_user(user: UserSchema):
    # user is a dictionary containing the user's data
    pass
```

**2. Validating Request Data**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields, ValidationError

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer(required=True)
    name = fields.String(required=True)
    age = fields.Integer(required=True)

@app.post("/users")
async def create_user(user: UserSchema):
    try:
        user.validate()
        # user data is valid
    except ValidationError as err:
        # user data is invalid
        pass
```

**3. Parsing Multipart/Form Data**

```python
from fastapi import FastAPI, File, UploadFile
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()
    profile_image = fields.Field()

@app.post("/users")
async def create_user(user: UserSchema, profile_image: UploadFile = File(...)):
    # user data and profile image are available
    pass
```

**4. Customizing Serialization and Deserialization**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields, post_load

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()
    age = fields.Integer()

    @post_load
    def make_user(self, data, **kwargs):
        # custom post-deserialization hook
        return User(**data)

class User:
    def __init__(self, id, name, age):
        self.id = id
        self.name = name
        self.age = age

@app.post("/users")
async def create_user(user: UserSchema):
    # user is a User object
    pass
```

**5. Using Nested Schemas**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields

app = FastAPI()

class AddressSchema(Schema):
    street = fields.String()
    city = fields.String()
    state = fields.String()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()
    age = fields.Integer()
    address = fields.Nested(AddressSchema)

@app.post("/users")
async def create_user(user: UserSchema):
    # user data and address data are available
    pass
```

**6. Using Schemas with the Pydantic Model**

```python
from fastapi import FastAPI
from pydantic import BaseModel, Field
from marshmallow import Schema, fields

app = FastAPI()

class AddressSchema(Schema):
    street = fields.String()
    city = fields.String()
    state = fields.String()

class UserSchema(Schema):
    id = fields.Integer(required=True)
    name = fields.String(required=True)
    age = fields.Integer(required=True)
    address = fields.Nested(AddressSchema)

class User(BaseModel):
    id: int = Field(schema=UserSchema.id)
    name: str = Field(schema=UserSchema.name)
    age: int = Field(schema=UserSchema.age)
    address: AddressSchema = Field(schema=UserSchema.address)

@app.post("/users")
async def create_user(user: User):
    # user data and address data are available in the Pydantic model
    pass
```

**7. Using Schemas with the Alembic Migration Tool**

```python
from alembic import op
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    age = Column(Integer, nullable=False)

class Address(Base):
    __tablename__ = "addresses"

    id = Column(Integer, primary_key=True)
    street = Column(String, nullable=False)
    city = Column(String, nullable=False)
    state = Column(String, nullable=False)

class UserSchema(Schema):
    id = fields.Integer(required=True)
    name = fields.String(required=True)
    age = fields.Integer(required=True)
    address = fields.Nested(AddressSchema)

def upgrade():
    op.add_column("users", Column("address_id", Integer, nullable=True))
    op.create_foreign_key("fk_user_address", "users", "addresses", ["address_id"], ["id"])

def downgrade():
    op.drop_constraint("fk_user_address", "users")
    op.drop_column("users", "address_id")
```

**8. Serializing and Deserializing Data with the `load_from` and `dump_to` Methods**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()
    age = fields.Integer()

user_data = {"id": 1, "name": "John", "age": 30}

# Serializing
user_json = UserSchema().dump(user_data)

# Deserializing
user_dict = UserSchema().load(user_json)
```

**9. Using the `exclude` Option to Exclude Fields from Serialization**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()
    age = fields.Integer()
    password = fields.String(exclude=True)

user_data = {"id": 1, "name": "John", "age": 30, "password": "secret"}

# Serializing
user_json = UserSchema().dump(user_data)
# Output: {"id": 1, "name": "John", "age": 30}
```

**10. Using the `only` Option to Include Only Specific Fields in Serialization**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()
    age = fields.Integer()

user_data = {"id": 1, "name": "John", "age": 30}

# Serializing
user_json = UserSchema(only=("id", "name")).dump(user_data)
# Output: {"id": 1, "name": "John"}
```

**11. Using the `required` Option to Validate Required Fields**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields, ValidationError

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer(required=True)
    name = fields.String(required=True)
    age = fields.Integer(required=True)

try:
    # Validating with missing data
    UserSchema().load({"id": 1, "name": "John"})
    # Raises ValidationError
except ValidationError as err:
    # Handle the validation error
    pass
```

**12. Using the `validate` Method to Manually Validate Data**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()
    age = fields.Integer()

user_data = {"id": 1, "name": "John", "age": 30}

# Manual validation
errors = UserSchema().validate(user_data)
if errors:
    # Handle the validation errors
    pass
```

**13. Using Schemas with the FastAPI Security Middleware**

```python
from fastapi import FastAPI, Security
from fastapi.security.api_key import APIKeyHeader
from marshmallow import Schema, fields

app = FastAPI()

api_key_header = APIKeyHeader(name="api_key", auto_error=False)

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.post("/users", dependencies=[Security(api_key_header)])
async def create_user(user: UserSchema):
    # user data is deserialized and validated
    pass
```

**14. Using Schemas with the FastAPI Routing**

```python
from fastapi import FastAPI, Path, Query
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.get("/users/{user_id}", response_model=UserSchema)
async def get_user(user_id: int = Path(...)):
    # user_id is validated and converted to an integer
    pass

@app.get("/users", response_model=list[UserSchema])
async def get_users(name: str = Query(...)):
    # name is validated and converted to a string
    pass
```

**15. Using Schemas with the FastAPI Dependency Injection**

```python
from fastapi import FastAPI, Depends
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

async def get_user(user_id: int) -> UserSchema:
    # Fetching user data from the database
    return UserSchema().load({"id": user_id, "name": "John"})

@app.get("/users/{user_id}")
async def get_user_endpoint(user: UserSchema = Depends(get_user)):
    # user data is deserialized and validated
    pass
```

**16. Using Schemas with the FastAPI Form Data**

```python
from fastapi import FastAPI, Form
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.post("/users")
async def create_user(user: UserSchema = Form(...)):
    # user data is deserialized and validated
    pass
```

**17. Using Schemas with the FastAPI File Upload**

```python
from fastapi import FastAPI, File, UploadFile
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()
    profile_image = fields.Field()

@app.post("/users")
async def create_user(user: UserSchema, profile_image: UploadFile = File(...)):
    # user data and profile image are deserialized and validated
    pass
```

**18. Using Schemas with the FastAPI Rate Limiting**

```python
from fastapi import FastAPI, Request
from fastapi.responses import HTTPException
from marshmallow import Schema, fields

app = FastAPI()

class RateLimitSchema(Schema):
    requests = fields.Integer()
    limit = fields.Integer()

@app.middleware("http")
async def check_rate_limit(request: Request, call_next):
    # Fetching rate limit data from the database
    rate_limit = RateLimitSchema().load({"requests": 10, "limit": 100})

    # Checking if the request exceeds the rate limit
    if rate_limit.requests >= rate_limit.limit:
        raise HTTPException(429, "Rate limit exceeded")

    # Allowing the request to proceed
    response = await call_next(request)
    return response
```

**19. Using Schemas with the FastAPI OpenAPI Documentation**

```python
from fastapi import FastAPI, OpenAPISchema
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

app.openapi_schema = OpenAPISchema(
    title="My API",
    description="This is my awesome API",
    schemas={
        "User": UserSchema
    }
)
```

**20. Using Schemas with the FastAPI Redoc Documentation**

```python
from fastapi import FastAPI, Redoc, RedocSettings
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

app.redoc = Redoc(
    title="My API",
    description="This is my awesome API",
    schemas={
        "User": UserSchema
    }
)
```

**21. Using Schemas with the FastAPI Swagger UI Documentation**

```python
from fastapi import FastAPI, SwaggerUI, SwaggerUIRouter
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

swagger_ui_router = SwaggerUIRouter(app=app)
app.include_router(swagger_ui_router, prefix="/docs")
```

**22. Using Schemas with the FastAPI Alembic Migrations**

```python
from alembic import op
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    age = Column(Integer, nullable=False)

class Address(Base):
    __tablename__ = "addresses"

    id = Column(Integer, primary_key=True)
    street = Column(String, nullable=False)
    city = Column(String, nullable=False)
    state = Column(String, nullable=False)

class UserSchema(Schema):
    id = fields.Integer(required=True)
    name = fields.String(required=True)
    age = fields.Integer(required=True)
    address = fields.Nested(AddressSchema)

def upgrade():
    op.add_column("users", Column("address_id", Integer, nullable=True))
    op.create_foreign_key("fk_user_address", "users", "addresses", ["address_id"], ["id"])

def downgrade():
    op.drop_constraint("fk_user_address", "users")
    op.drop_column("users", "address_id")
```

**23. Using Schemas with the FastAPI Celery Tasks**

```python
from celery import Celery
from marshmallow import Schema, fields

app = FastAPI()
celery = Celery(app.title)

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@celery.task
def create_user(user_data):
    # user_data is deserialized and validated
    pass
```

**24. Using Schemas with the FastAPI Dependency Injection**

```python
from fastapi import Depends
from marshmallow import Schema, fields

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

async def get_user(user_id: int) -> UserSchema:
    # Fetching user data from the database
    return UserSchema().load({"id": user_id, "name": "John"})

@app.get("/users/{user_id}")
async def get_user_endpoint(user: UserSchema = Depends(get_user)):
    # user data is deserialized and validated
    pass
```

**25. Using Schemas with the FastAPI Middleware**

```python
from fastapi import FastAPI, Request
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.middleware("http")
async def check_authentication(request: Request, call_next):
    # Fetching authentication data from the request
    auth_data = UserSchema().load({"id": 1, "name": "John"})

    # Checking if the user is authenticated
    if not auth_data:
        raise HTTPException(401, "Unauthorized")

    # Allowing the request to proceed
    response = await call_next(request)
    return response
```

**26. Using Schemas with the FastAPI Pydantic Models**

```python
from fastapi import FastAPI, Body
from marshmallow import Schema, fields
from pydantic import BaseModel

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

class User(BaseModel):
    id: int
    name: str

@app.post("/users")
async def create_user(user: User = Body(...)):
    # user data is deserialized and validated
    pass
```

**27. Using Schemas with the FastAPI Request Body**

```python
from fastapi import FastAPI, Request
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.post("/users")
async def create_user(request: Request):
    # Fetching user data from the request body
    user_data = await request.json()
    user = UserSchema().load(user_data)

    # user data is deserialized and validated
    pass
```

**28. Using Schemas with the FastAPI Request Query Parameters**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.get("/users")
async def get_users(name: str = Query(...)):
    # name is deserialized and validated
    pass
```

**29. Using Schemas with the FastAPI Path Parameters**

```python
from fastapi import FastAPI
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.get("/users/{user_id}")
async def get_user(user_id: int = Path(...)):
    # user_id is deserialized and validated
    pass
```

**30. Using Schemas with the FastAPI Form Data**

```python
from fastapi import FastAPI, Form
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.post("/users")
async def create_user(user: UserSchema = Form(...)):
    # user data is deserialized and validated
    pass
```

**31. Using Schemas with the FastAPI WebSocket**

```python
from fastapi import FastAPI, WebSocket
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.websocket("/users")
async def websocket_handler(websocket: WebSocket):
    # Fetching user data from the WebSocket
    user_data = await websocket.receive_json()
    user = UserSchema().load(user_data)

    # user data is deserialized and validated
    pass
```

**32. Using Schemas with the FastAPI HTTPX Client**

```python
from fastapi import FastAPI, HTTPXClient
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

client = HTTPXClient(app)

async def get_user(user_id: int):
    # Fetching user data from the remote API
    response = await client.get(f"/users/{user_id}")
    user = UserSchema().load(response.json())

    # user data is deserialized and validated
    pass
```

**33. Using Schemas with the FastAPI Alembic Migrations**

```python
from alembic import op
from sqlalchemy import Column, Integer, String
from sqlalchemy.ext.declarative import declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = "users"

    id = Column(Integer, primary_key=True)
    name = Column(String, nullable=False)
    age = Column(Integer, nullable=False)

class Address(Base):
    __tablename__ = "addresses"

    id = Column(Integer, primary_key=True)
    street = Column(String, nullable=False)
    city = Column(String, nullable=False)
    state = Column(String, nullable=False)

class UserSchema(Schema):
    id = fields.Integer(required=True)
    name = fields.String(required=True)
    age = fields.Integer(required=True)
    address = fields.Nested(AddressSchema)

def upgrade():
    op.add_column("users", Column("address_id", Integer, nullable=True))
    op.create_foreign_key("fk_user_address", "users", "addresses", ["address_id"], ["id"])

def downgrade():
    op.drop_constraint("fk_user_address", "users")
    op.drop_column("users", "address_id")
```

**34. Using Schemas with the FastAPI Jinja2 Templates**

```python
from fastapi import FastAPI, TemplateResponse
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.get("/users")
async def get_users():
    # Fetching user data from the database
    users = [
        {"id": 1, "name": "John"},
        {"id": 2, "name": "Jane"}
    ]

    # Serializing the user data
    user_schemas = UserSchema(many=True).dump(users)

    # Rendering the Jinja2 template
    return TemplateResponse("users.html", {"users": user_schemas})
```

**35. Using Schemas with the FastAPI Starlette Responses**

```python
from fastapi import FastAPI, Response
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.get("/users")
async def get_users():
    # Fetching user data from the database
    users = [
        {"id": 1, "name": "John"},
        {"id": 2, "name": "Jane"}
    ]

    # Serializing the user data
    user_schemas = UserSchema(many=True).dump(users)

    # Returning the Starlette Response
    return Response(content=user_schemas, media_type="application/json")
```

**36. Using Schemas with the FastAPI Background Tasks**

```python
from fastapi import FastAPI, BackgroundTasks
from marshmallow import Schema, fields

app = FastAPI()

class UserSchema(Schema):
    id = fields.Integer()
    name = fields.String()

@app.post("/users")
async def create_user(background_tasks: BackgroundTasks, user: UserSchema):
    # Fetching user data from the database
    users = [
        {"id": 1, "name": "John"},
        {"id": 2, "name": "Jane"}
    ]

    # Serializing the user data
    user_schemas = UserSchema(many=True).dump(users)

    # Adding a background task to send email notifications
    background_tasks.add_task(send_email_notifications, user_schemas)

    # Returning the response
    return user_schemas
```