Module aws_lambda_powertools.utilities.parser.models.apigw

Expand source code
from datetime import datetime
from typing import Any, Dict, List, Optional

from pydantic import BaseModel, root_validator
from pydantic.networks import IPvAnyNetwork

from ..types import Literal


class ApiGatewayUserCertValidity(BaseModel):
    notBefore: str
    notAfter: str


class ApiGatewayUserCert(BaseModel):
    clientCertPem: str
    subjectDN: str
    issuerDN: str
    serialNumber: str
    validity: ApiGatewayUserCertValidity


class APIGatewayEventIdentity(BaseModel):
    accessKey: Optional[str]
    accountId: Optional[str]
    apiKey: Optional[str]
    apiKeyId: Optional[str]
    caller: Optional[str]
    cognitoAuthenticationProvider: Optional[str]
    cognitoAuthenticationType: Optional[str]
    cognitoIdentityId: Optional[str]
    cognitoIdentityPoolId: Optional[str]
    principalOrgId: Optional[str]
    sourceIp: IPvAnyNetwork
    user: Optional[str]
    userAgent: Optional[str]
    userArn: Optional[str]
    clientCert: Optional[ApiGatewayUserCert]


class APIGatewayEventAuthorizer(BaseModel):
    claims: Optional[Dict[str, Any]]
    scopes: Optional[List[str]]


class APIGatewayEventRequestContext(BaseModel):
    accountId: str
    apiId: str
    authorizer: Optional[APIGatewayEventAuthorizer]
    stage: str
    protocol: str
    identity: APIGatewayEventIdentity
    requestId: str
    requestTime: str
    requestTimeEpoch: datetime
    resourceId: Optional[str]
    resourcePath: str
    domainName: Optional[str]
    domainPrefix: Optional[str]
    extendedRequestId: Optional[str]
    httpMethod: Literal["DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT"]
    path: str
    connectedAt: Optional[datetime]
    connectionId: Optional[str]
    eventType: Optional[Literal["CONNECT", "MESSAGE", "DISCONNECT"]]
    messageDirection: Optional[str]
    messageId: Optional[str]
    routeKey: Optional[str]
    operationName: Optional[str]


class APIGatewayProxyEventModel(BaseModel):
    version: Optional[str]
    resource: str
    path: str
    httpMethod: Literal["DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT"]
    headers: Dict[str, str]
    multiValueHeaders: Dict[str, List[str]]
    queryStringParameters: Optional[Dict[str, str]]
    multiValueQueryStringParameters: Optional[Dict[str, List[str]]]
    requestContext: APIGatewayEventRequestContext
    pathParameters: Optional[Dict[str, str]]
    stageVariables: Optional[Dict[str, str]]
    isBase64Encoded: bool
    body: str

    @root_validator()
    def check_message_id(cls, values):
        message_id, event_type = values.get("messageId"), values.get("eventType")
        if message_id is not None and event_type != "MESSAGE":
            raise TypeError("messageId is available only when the `eventType` is `MESSAGE`")
        return values

Classes

class APIGatewayEventAuthorizer (**data: Any)

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code
class APIGatewayEventAuthorizer(BaseModel):
    claims: Optional[Dict[str, Any]]
    scopes: Optional[List[str]]

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var claims : Union[Dict[str, Any], NoneType]
var scopes : Union[List[str], NoneType]
class APIGatewayEventIdentity (**data: Any)

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code
class APIGatewayEventIdentity(BaseModel):
    accessKey: Optional[str]
    accountId: Optional[str]
    apiKey: Optional[str]
    apiKeyId: Optional[str]
    caller: Optional[str]
    cognitoAuthenticationProvider: Optional[str]
    cognitoAuthenticationType: Optional[str]
    cognitoIdentityId: Optional[str]
    cognitoIdentityPoolId: Optional[str]
    principalOrgId: Optional[str]
    sourceIp: IPvAnyNetwork
    user: Optional[str]
    userAgent: Optional[str]
    userArn: Optional[str]
    clientCert: Optional[ApiGatewayUserCert]

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var accessKey : Union[str, NoneType]
var accountId : Union[str, NoneType]
var apiKey : Union[str, NoneType]
var apiKeyId : Union[str, NoneType]
var caller : Union[str, NoneType]
var clientCert : Union[ApiGatewayUserCert, NoneType]
var cognitoAuthenticationProvider : Union[str, NoneType]
var cognitoAuthenticationType : Union[str, NoneType]
var cognitoIdentityId : Union[str, NoneType]
var cognitoIdentityPoolId : Union[str, NoneType]
var principalOrgId : Union[str, NoneType]
var sourceIp : pydantic.networks.IPvAnyNetwork
var user : Union[str, NoneType]
var userAgent : Union[str, NoneType]
var userArn : Union[str, NoneType]
class APIGatewayEventRequestContext (**data: Any)

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code
class APIGatewayEventRequestContext(BaseModel):
    accountId: str
    apiId: str
    authorizer: Optional[APIGatewayEventAuthorizer]
    stage: str
    protocol: str
    identity: APIGatewayEventIdentity
    requestId: str
    requestTime: str
    requestTimeEpoch: datetime
    resourceId: Optional[str]
    resourcePath: str
    domainName: Optional[str]
    domainPrefix: Optional[str]
    extendedRequestId: Optional[str]
    httpMethod: Literal["DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT"]
    path: str
    connectedAt: Optional[datetime]
    connectionId: Optional[str]
    eventType: Optional[Literal["CONNECT", "MESSAGE", "DISCONNECT"]]
    messageDirection: Optional[str]
    messageId: Optional[str]
    routeKey: Optional[str]
    operationName: Optional[str]

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var accountId : str
var apiId : str
var authorizer : Union[APIGatewayEventAuthorizer, NoneType]
var connectedAt : Union[datetime.datetime, NoneType]
var connectionId : Union[str, NoneType]
var domainName : Union[str, NoneType]
var domainPrefix : Union[str, NoneType]
var eventType : Union[Literal['CONNECT', 'MESSAGE', 'DISCONNECT'], NoneType]
var extendedRequestId : Union[str, NoneType]
var httpMethod : Literal['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PATCH', 'POST', 'PUT']
var identityAPIGatewayEventIdentity
var messageDirection : Union[str, NoneType]
var messageId : Union[str, NoneType]
var operationName : Union[str, NoneType]
var path : str
var protocol : str
var requestId : str
var requestTime : str
var requestTimeEpoch : datetime.datetime
var resourceId : Union[str, NoneType]
var resourcePath : str
var routeKey : Union[str, NoneType]
var stage : str
class APIGatewayProxyEventModel (**data: Any)

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code
class APIGatewayProxyEventModel(BaseModel):
    version: Optional[str]
    resource: str
    path: str
    httpMethod: Literal["DELETE", "GET", "HEAD", "OPTIONS", "PATCH", "POST", "PUT"]
    headers: Dict[str, str]
    multiValueHeaders: Dict[str, List[str]]
    queryStringParameters: Optional[Dict[str, str]]
    multiValueQueryStringParameters: Optional[Dict[str, List[str]]]
    requestContext: APIGatewayEventRequestContext
    pathParameters: Optional[Dict[str, str]]
    stageVariables: Optional[Dict[str, str]]
    isBase64Encoded: bool
    body: str

    @root_validator()
    def check_message_id(cls, values):
        message_id, event_type = values.get("messageId"), values.get("eventType")
        if message_id is not None and event_type != "MESSAGE":
            raise TypeError("messageId is available only when the `eventType` is `MESSAGE`")
        return values

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var body : str
var headers : Dict[str, str]
var httpMethod : Literal['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PATCH', 'POST', 'PUT']
var isBase64Encoded : bool
var multiValueHeaders : Dict[str, List[str]]
var multiValueQueryStringParameters : Union[Dict[str, List[str]], NoneType]
var path : str
var pathParameters : Union[Dict[str, str], NoneType]
var queryStringParameters : Union[Dict[str, str], NoneType]
var requestContextAPIGatewayEventRequestContext
var resource : str
var stageVariables : Union[Dict[str, str], NoneType]
var version : Union[str, NoneType]

Static methods

def check_message_id(values)
Expand source code
@root_validator()
def check_message_id(cls, values):
    message_id, event_type = values.get("messageId"), values.get("eventType")
    if message_id is not None and event_type != "MESSAGE":
        raise TypeError("messageId is available only when the `eventType` is `MESSAGE`")
    return values
class ApiGatewayUserCert (**data: Any)

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code
class ApiGatewayUserCert(BaseModel):
    clientCertPem: str
    subjectDN: str
    issuerDN: str
    serialNumber: str
    validity: ApiGatewayUserCertValidity

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var clientCertPem : str
var issuerDN : str
var serialNumber : str
var subjectDN : str
var validityApiGatewayUserCertValidity
class ApiGatewayUserCertValidity (**data: Any)

Create a new model by parsing and validating input data from keyword arguments.

Raises ValidationError if the input data cannot be parsed to form a valid model.

Expand source code
class ApiGatewayUserCertValidity(BaseModel):
    notBefore: str
    notAfter: str

Ancestors

  • pydantic.main.BaseModel
  • pydantic.utils.Representation

Class variables

var notAfter : str
var notBefore : str