Module aws_lambda_powertools.utilities.parser.models.s3
Expand source code
from datetime import datetime
from typing import List, Optional
from pydantic import BaseModel, root_validator
from pydantic.fields import Field
from pydantic.networks import IPvAnyNetwork
from pydantic.types import NonNegativeFloat
from aws_lambda_powertools.utilities.parser.types import Literal
from .event_bridge import EventBridgeModel
class S3EventRecordGlacierRestoreEventData(BaseModel):
lifecycleRestorationExpiryTime: datetime
lifecycleRestoreStorageClass: str
class S3EventRecordGlacierEventData(BaseModel):
restoreEventData: S3EventRecordGlacierRestoreEventData
class S3Identity(BaseModel):
principalId: str
class S3RequestParameters(BaseModel):
sourceIPAddress: IPvAnyNetwork
class S3ResponseElements(BaseModel):
x_amz_request_id: str = Field(None, alias="x-amz-request-id")
x_amz_id_2: str = Field(None, alias="x-amz-id-2")
class S3OwnerIdentify(BaseModel):
principalId: str
class S3Bucket(BaseModel):
name: str
ownerIdentity: S3OwnerIdentify
arn: str
class S3Object(BaseModel):
key: str
size: Optional[NonNegativeFloat] = None
eTag: Optional[str] = None
sequencer: str
versionId: Optional[str] = None
class S3Message(BaseModel):
s3SchemaVersion: str
configurationId: str
bucket: S3Bucket
object: S3Object # noqa: A003,VNE003
class S3EventNotificationObjectModel(BaseModel):
key: str
size: Optional[NonNegativeFloat] = None
etag: str
version_id: str = Field(None, alias="version-id")
sequencer: Optional[str] = None
class S3EventNotificationEventBridgeBucketModel(BaseModel):
name: str
class S3EventNotificationEventBridgeDetailModel(BaseModel):
version: str
bucket: S3EventNotificationEventBridgeBucketModel
object: S3EventNotificationObjectModel # noqa: A003,VNE003
request_id: str = Field(None, alias="request-id")
requester: str
source_ip_address: str = Field(None, alias="source-ip-address")
reason: Optional[str] = None
deletion_type: Optional[str] = Field(None, alias="deletion-type")
restore_expiry_time: Optional[str] = Field(None, alias="restore-expiry-time")
source_storage_class: Optional[str] = Field(None, alias="source-storage-class")
destination_storage_class: Optional[str] = Field(None, alias="destination-storage-class")
destination_access_tier: Optional[str] = Field(None, alias="destination-access-tier")
class S3EventNotificationEventBridgeModel(EventBridgeModel):
detail: S3EventNotificationEventBridgeDetailModel
class S3RecordModel(BaseModel):
eventVersion: str
eventSource: Literal["aws:s3"]
awsRegion: str
eventTime: datetime
eventName: str
userIdentity: S3Identity
requestParameters: S3RequestParameters
responseElements: S3ResponseElements
s3: S3Message
glacierEventData: Optional[S3EventRecordGlacierEventData] = None
@root_validator(allow_reuse=True, skip_on_failure=True)
def validate_s3_object(cls, values):
event_name = values.get("eventName")
s3_object = values.get("s3").object
if "ObjectRemoved" not in event_name:
if s3_object.size is None or s3_object.eTag is None:
raise ValueError("S3Object.size and S3Object.eTag are required for non-ObjectRemoved events")
return values
class S3Model(BaseModel):
Records: List[S3RecordModel]
Classes
class S3Bucket (**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 S3Bucket(BaseModel): name: str ownerIdentity: S3OwnerIdentify arn: str
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var arn : str
var name : str
var ownerIdentity : S3OwnerIdentify
class S3EventNotificationEventBridgeBucketModel (**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 S3EventNotificationEventBridgeBucketModel(BaseModel): name: str
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var name : str
class S3EventNotificationEventBridgeDetailModel (**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 S3EventNotificationEventBridgeDetailModel(BaseModel): version: str bucket: S3EventNotificationEventBridgeBucketModel object: S3EventNotificationObjectModel # noqa: A003,VNE003 request_id: str = Field(None, alias="request-id") requester: str source_ip_address: str = Field(None, alias="source-ip-address") reason: Optional[str] = None deletion_type: Optional[str] = Field(None, alias="deletion-type") restore_expiry_time: Optional[str] = Field(None, alias="restore-expiry-time") source_storage_class: Optional[str] = Field(None, alias="source-storage-class") destination_storage_class: Optional[str] = Field(None, alias="destination-storage-class") destination_access_tier: Optional[str] = Field(None, alias="destination-access-tier")
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var bucket : S3EventNotificationEventBridgeBucketModel
var deletion_type : Optional[str]
var destination_access_tier : Optional[str]
var destination_storage_class : Optional[str]
var object : S3EventNotificationObjectModel
var reason : Optional[str]
var request_id : str
var requester : str
var restore_expiry_time : Optional[str]
var source_ip_address : str
var source_storage_class : Optional[str]
var version : str
class S3EventNotificationEventBridgeModel (**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 S3EventNotificationEventBridgeModel(EventBridgeModel): detail: S3EventNotificationEventBridgeDetailModel
Ancestors
- EventBridgeModel
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var detail : S3EventNotificationEventBridgeDetailModel
class S3EventNotificationObjectModel (**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 S3EventNotificationObjectModel(BaseModel): key: str size: Optional[NonNegativeFloat] = None etag: str version_id: str = Field(None, alias="version-id") sequencer: Optional[str] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var etag : str
var key : str
var sequencer : Optional[str]
var size : Optional[pydantic.types.NonNegativeFloat]
var version_id : str
class S3EventRecordGlacierEventData (**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 S3EventRecordGlacierEventData(BaseModel): restoreEventData: S3EventRecordGlacierRestoreEventData
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var restoreEventData : S3EventRecordGlacierRestoreEventData
class S3EventRecordGlacierRestoreEventData (**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 S3EventRecordGlacierRestoreEventData(BaseModel): lifecycleRestorationExpiryTime: datetime lifecycleRestoreStorageClass: str
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var lifecycleRestorationExpiryTime : datetime.datetime
var lifecycleRestoreStorageClass : str
class S3Identity (**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 S3Identity(BaseModel): principalId: str
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var principalId : str
class S3Message (**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 S3Message(BaseModel): s3SchemaVersion: str configurationId: str bucket: S3Bucket object: S3Object # noqa: A003,VNE003
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var bucket : S3Bucket
var configurationId : str
var object : S3Object
var s3SchemaVersion : str
class S3Model (**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 S3Model(BaseModel): Records: List[S3RecordModel]
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var Records : List[S3RecordModel]
class S3Object (**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 S3Object(BaseModel): key: str size: Optional[NonNegativeFloat] = None eTag: Optional[str] = None sequencer: str versionId: Optional[str] = None
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var eTag : Optional[str]
var key : str
var sequencer : str
var size : Optional[pydantic.types.NonNegativeFloat]
var versionId : Optional[str]
class S3OwnerIdentify (**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 S3OwnerIdentify(BaseModel): principalId: str
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var principalId : str
class S3RecordModel (**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 S3RecordModel(BaseModel): eventVersion: str eventSource: Literal["aws:s3"] awsRegion: str eventTime: datetime eventName: str userIdentity: S3Identity requestParameters: S3RequestParameters responseElements: S3ResponseElements s3: S3Message glacierEventData: Optional[S3EventRecordGlacierEventData] = None @root_validator(allow_reuse=True, skip_on_failure=True) def validate_s3_object(cls, values): event_name = values.get("eventName") s3_object = values.get("s3").object if "ObjectRemoved" not in event_name: if s3_object.size is None or s3_object.eTag is None: raise ValueError("S3Object.size and S3Object.eTag are required for non-ObjectRemoved events") return values
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var awsRegion : str
var eventName : str
var eventSource : Literal['aws:s3']
var eventTime : datetime.datetime
var eventVersion : str
var glacierEventData : Optional[S3EventRecordGlacierEventData]
var requestParameters : S3RequestParameters
var responseElements : S3ResponseElements
var s3 : S3Message
var userIdentity : S3Identity
Static methods
def validate_s3_object(values)
-
Expand source code
@root_validator(allow_reuse=True, skip_on_failure=True) def validate_s3_object(cls, values): event_name = values.get("eventName") s3_object = values.get("s3").object if "ObjectRemoved" not in event_name: if s3_object.size is None or s3_object.eTag is None: raise ValueError("S3Object.size and S3Object.eTag are required for non-ObjectRemoved events") return values
class S3RequestParameters (**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 S3RequestParameters(BaseModel): sourceIPAddress: IPvAnyNetwork
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var sourceIPAddress : pydantic.networks.IPvAnyNetwork
class S3ResponseElements (**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 S3ResponseElements(BaseModel): x_amz_request_id: str = Field(None, alias="x-amz-request-id") x_amz_id_2: str = Field(None, alias="x-amz-id-2")
Ancestors
- pydantic.main.BaseModel
- pydantic.utils.Representation
Class variables
var x_amz_id_2 : str
var x_amz_request_id : str