Module aws_lambda_powertools.utilities.parser.models.apigw
Expand source code
from datetime import datetime
from typing import Any, Dict, List, Optional, Type, Union
from pydantic import BaseModel, root_validator
from pydantic.networks import IPvAnyNetwork
from aws_lambda_powertools.utilities.parser.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]
# see #1562, temp workaround until API Gateway fixes it the Test button payload
# removing it will not be considered a regression in the future
sourceIp: Union[IPvAnyNetwork, Literal["test-invoke-source-ip"]]
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]
@root_validator(allow_reuse=True)
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 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: Optional[Union[str, Type[BaseModel]]]
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 : Optional[Dict[str, Any]]
var scopes : Optional[List[str]]
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] # see #1562, temp workaround until API Gateway fixes it the Test button payload # removing it will not be considered a regression in the future sourceIp: Union[IPvAnyNetwork, Literal["test-invoke-source-ip"]] user: Optional[str] userAgent: Optional[str] userArn: Optional[str] clientCert: Optional[ApiGatewayUserCert]
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var accessKey : Optional[str]
var accountId : Optional[str]
var apiKey : Optional[str]
var apiKeyId : Optional[str]
var caller : Optional[str]
var clientCert : Optional[ApiGatewayUserCert]
var cognitoAuthenticationProvider : Optional[str]
var cognitoAuthenticationType : Optional[str]
var cognitoIdentityId : Optional[str]
var cognitoIdentityPoolId : Optional[str]
var principalOrgId : Optional[str]
var sourceIp : Union[pydantic.networks.IPvAnyNetwork, Literal['test-invoke-source-ip']]
var user : Optional[str]
var userAgent : Optional[str]
var userArn : Optional[str]
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] @root_validator(allow_reuse=True) 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 accountId : str
var apiId : str
var connectedAt : Optional[datetime.datetime]
var connectionId : Optional[str]
var domainName : Optional[str]
var domainPrefix : Optional[str]
var eventType : Optional[Literal['CONNECT', 'MESSAGE', 'DISCONNECT']]
var extendedRequestId : Optional[str]
var httpMethod : Literal['DELETE', 'GET', 'HEAD', 'OPTIONS', 'PATCH', 'POST', 'PUT']
var identity : APIGatewayEventIdentity
var messageDirection : Optional[str]
var messageId : Optional[str]
var operationName : Optional[str]
var path : str
var protocol : str
var requestId : str
var requestTime : str
var requestTimeEpoch : datetime.datetime
var resourceId : Optional[str]
var resourcePath : str
var routeKey : Optional[str]
var stage : str
Static methods
def check_message_id(values)
-
Expand source code
@root_validator(allow_reuse=True) 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 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: Optional[Union[str, Type[BaseModel]]]
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var body : Union[str, Type[pydantic.main.BaseModel], ForwardRef(None)]
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 : Optional[Dict[str, List[str]]]
var path : str
var pathParameters : Optional[Dict[str, str]]
var queryStringParameters : Optional[Dict[str, str]]
var requestContext : APIGatewayEventRequestContext
var resource : str
var stageVariables : Optional[Dict[str, str]]
var version : Optional[str]
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 validity : ApiGatewayUserCertValidity
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