bi_etl.boto3_helper.s3 module

Created on March 28, 2022

@author:

class bi_etl.boto3_helper.s3.Boto3_S3(*, user_id: str | None = None, password_source: PasswordSource, <function _check_password_source at 0x7f459f4075e0>] | None=None, raw_password: str | None = None, keyring_section: str | None = None, keepass_config: str | None = 'keepass', keepass: KeepassConfig | None = None, keepass_group: str | None = None, keepass_title: str | None = None, validate_password_on_load: bool = True, region_name: str | None = None, iam_role: str | None = None, bucket_name: str, key: str | None = None, treat_as_folder: bool = False)[source]

Bases: S3_Bucket

class CompareResult(*values)

Bases: Enum

Enumeration of file comparison results between S3 and local files.

DIFFERENT_SIZE

Files have different sizes.

Type:: auto

LOCAL_NEWER

Local file has a newer modification timestamp.

Type:: auto

SAME_TIMES

Files have identical modification timestamps.

Type:: auto

LOCAL_OLDER

Local file has an older modification timestamp.

Type:: auto

DIFFERENT_SIZE = 1

LOCAL_NEWER = 2

LOCAL_OLDER = 4

SAME_TIMES = 3

class OverwriteModes(*values)

Bases: Enum

Enumeration of file overwrite behavior modes for S3 operations.

ALWAYS_OVERWRITE

Always overwrite the target file regardless of timestamps or sizes.

Type:: auto

OVERWRITE_OLDER

Only overwrite if source is newer or sizes differ.

Type:: auto

NEVER_OVERWRITE

Never overwrite existing files.

Type:: auto

ALWAYS_OVERWRITE = 1

NEVER_OVERWRITE = 3

OVERWRITE_OLDER = 2

__init__(**data: Any) → None

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.

Uses something other than self the first arg to allow “self” as a settable attribute

add_child(name: str, child_object: ConfigHierarchy): Set this configuration as a child in the hierarchy of another config. For any programmatically created config objects this is required so that the new object ‘knows’ where it lives in the hierarchy – most importantly so that it can find the hierarchies root object.

as_bucket() → S3_Bucket

Create a new S3_Bucket instance with only the bucket name.

Returns:: S3_Bucket instance without key, folder, or file_name attributes.
Return type:: S3_Bucket

assume_role()

bucket_name: The name of the S3 bucket

check_model()

property client: S3Client

Get the boto3 S3 client.

Returns:: The boto3 S3 client.
Return type:: S3Client

classmethod construct(_fields_set: set[str] | None = None, **values: Any) → Self

content_type() → str

Get the content type (MIME type) of the S3 object.

Returns:: The content type string (e.g., ‘text/plain’, ‘application/json’).
Return type:: str

Returns a copy of the model.

!!! warning “Deprecated”: This method is now deprecated; use model_copy instead.

If you need include or exclude, use:

`python {test="skip" lint="skip"} data = self.model_dump(include=include, exclude=exclude, round_trip=True) data = {**data, **(update or {})} copied = self.model_validate(data) `

Parameters:

include¶ – Optional set or mapping specifying which fields to include in the copied model.
exclude¶ – Optional set or mapping specifying which fields to exclude in the copied model.
update¶ – Optional dictionary of field-value pairs to override field values in the copied model.
deep¶ – If True, the values of fields that are Pydantic models will be deep-copied.

Returns:

A copy of the model with included, excluded and updated fields as specified.

copy_to(target: str | S3_Bucket_Key, version_id: str | None = None)

Copy this S3 object to another S3 location.

Parameters:

target¶ (str | S3_Bucket_Key) – Target S3 location (URI string or S3_Bucket_Key instance).
version_id¶ (str | None, optional) – Specific version ID to copy. If None, copies the latest version.

delete(key: str | PurePosixPath | None = None, version_id: str | None = None)

Delete an S3 object.

Parameters:

key¶ (str | PurePosixPath | None, optional) – S3 key to delete. If None, uses the instance key.
version_id¶ (str | None, optional) – Specific version ID to delete. If None, deletes the latest version.

delete_by_key(key: str | PurePosixPath, version_id: str | None = None)

Delete an S3 object by key.

Deprecated since version Use: delete() instead.

Parameters:

key¶ (str | PurePosixPath) – S3 key to delete.
version_id¶ (str | None, optional) – Specific version ID to delete.

dict(*, include: set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None = None, exclude: set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None = None, by_alias: bool = False, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False) → Dict[str, Any]

download_file(*, local_filename: str | Path, key: str | PurePosixPath | None = None, extra_args: dict | None = None, transfer_config: TransferConfig | None = None, create_parents: bool = True, overwrite_mode: OverwriteModes = OverwriteModes.OVERWRITE_OLDER)

Download an S3 object to a local file.

Parameters:

local_filename¶ (Path | str) – Path where the downloaded file will be saved.
key¶ (str | PurePosixPath | None, optional) – S3 key to download. If None, uses the instance key.
extra_args¶ (dict | None, optional) – Extra arguments to pass to boto3 download_file.
transfer_config¶ (TransferConfig | None, optional) – Transfer configuration for the download.
create_parents¶ (bool, optional) – Whether to create parent directories. Default is True.
overwrite_mode¶ (OverwriteModes, optional) – Controls overwrite behavior. Default is OVERWRITE_OLDER.

Raises:

S3ClientError – If the S3 object does not exist or download fails.
ValueError – If a non-blank key is not specified for the download.

download_file_from_s3(file_key: str, local_file_path: str, date_placeholder: str = None, date_pattern: str = '%Y-%m-%d') → str[source]

static download_file_list_from_s3(file_list: List[File_Entry])[source]

download_files(*, local_path: str | Path, key: str | PurePosixPath | None = None, treat_as_folder: bool | None = None, extra_args: dict | None = None, transfer_config: TransferConfig | None = None, create_parents: bool = True, overwrite_mode: OverwriteModes = OverwriteModes.OVERWRITE_OLDER) → Iterable[Path]

Download multiple S3 objects under a key prefix to a local directory.

Parameters:

local_path¶ (str | Path) – Local directory path where files will be downloaded.
key¶ (str | PurePosixPath | None, optional) – S3 key prefix. If None, uses the instance key.
treat_as_folder¶ (bool | None, optional) – Whether to treat the key as a folder prefix. If None, uses instance setting.
extra_args¶ (dict | None, optional) – Extra arguments to pass to boto3 download_file.
transfer_config¶ (TransferConfig | None, optional) – Transfer configuration for downloads.
create_parents¶ (bool, optional) – Whether to create parent directories. Default is True.
overwrite_mode¶ (OverwriteModes, optional) – Controls overwrite behavior. Default is OVERWRITE_OLDER.

Returns:

List of local file paths that were downloaded.

Return type:

Iterable[Path]

download_files_from_s3(key_prefixes: List[str], local_folder: str, changed_only=True, return_full_list=True) → List[File_Entry][source]

exists(key: str | PurePosixPath | None = None, version_id: str | None = None) → bool

Check if an S3 object exists.

Parameters:

key¶ (str | PurePosixPath | None, optional) – S3 key to check. If None, uses the instance key.
version_id¶ (str | None, optional) – Specific version ID to check. If None, checks the latest version.

Returns:

True if the object exists, False otherwise.

Return type:

bool

Raises:

ClientError – If there’s an error other than 404/NoSuchKey.

static filter_list_changed_vs_local(file_list: List[File_Entry]) → List[File_Entry][source]

find_objects(key: str | PurePosixPath | None = None) → BucketObjectsCollection

List S3 objects under a key prefix.

Deprecated since version Use: list_objects() instead.

Parameters:: key¶ (str | PurePosixPath | None, optional) – S3 key prefix.
Returns:: Collection of S3 ObjectSummary objects.
Return type:: BucketObjectsCollection

classmethod from_orm(obj: Any) → Self

full_item_name(item_name: str | None = None, delimiter: str = ' -> '): The fully qualified name of this config item in the config hierarchy.

get_boto3_bucket() → Bucket

Get the boto3 Bucket resource object.

Returns:: The boto3 Bucket resource.
Return type:: Bucket
Raises:: ClientError – If the bucket does not exist (error code NoSuchBucket).

get_bucket_region() → str

Get the region_name from the actual S3 bucket definition.

Note

This can differ from the region_name attribute specified when constructing this class for example via a config file. That region_name attribute is used for establishing the AWS session.

NOTE 2:: If you don’t specify a Region, when the bucket is created it will have used the US East (N. Virginia) Region (us-east-1). So this method defaults to that if it does not find a LocationConstraint.

get_bucket_region_name() → str

Get the AWS region name where the bucket is located.

Returns:: AWS region name (e.g., ‘us-east-1’, ‘us-west-2’).
Return type:: str

Notes

Buckets in Region us-east-1 have a LocationConstraint of null.

get_copy(copied_by: str = 'get_copy') → AWS_Session: Copy this configuration. Useful when you need to programmatically modify a configuration without modifying the original base configuration.

get_object(key: str | PurePosixPath | None = None, version_id: str | None = None) → Object | ObjectVersion

Get a boto3 Object or ObjectVersion resource with caching.

Parameters:

key¶ (str | PurePosixPath | None, optional) – S3 key. If None, uses the instance key.
version_id¶ (str | None, optional) – Specific version ID. If None, returns the latest version Object.

Returns:

The boto3 Object or ObjectVersion resource.

Return type:

Object | ObjectVersion

get_object_uncached(key: str | PurePosixPath | None = None, version_id: str | None = None) → Object | ObjectVersion

Get a boto3 Object or ObjectVersion resource without caching.

Parameters:

key¶ (str | PurePosixPath | None, optional) – S3 key. If None, uses the instance key.
version_id¶ (str | None, optional) – Specific version ID. If None, returns the latest version Object.

Returns:

The boto3 Object or ObjectVersion resource.

Return type:

Object | ObjectVersion

get_password() → str: Get the password for this resource. password_source controls where it looks for the password. If that is None, then the root level passwords container is checked for password_source value.

get_request_v4_authorizer(service: str = None) → requests.auth.AuthBase

get_s3_bucket_key(key: str | PurePosixPath | None = None) → S3_Bucket_Key

Get an S3_Bucket_Key instance for the specified key.

Parameters:: key¶ (str | PurePosixPath | None, optional) – S3 key. If None, uses the instance key.
Returns:: S3_Bucket_Key instance representing the key.
Return type:: S3_Bucket_Key

get_secrets_manager()

get_service_client(service: str)

get_service_resource(service: str)

get_session_expiry()

get_ssm()

property has_session: bool

iam_role

is_file()

Check if this S3 object represents a file (not a directory marker).

Returns:: True if it’s a file, False if it’s a directory marker or doesn’t exist.
Return type:: bool

Notes

S3 doesn’t have true directories, but some tools create directory markers with content type ‘application/x-directory’.

is_relative_to(other: S3_Bucket)

Check if this S3 path is relative to another.

Parameters:: other¶ (S3_Bucket) – Another S3_Bucket instance to compare with.
Returns:: True if this path is relative to the other path, False otherwise.
Return type:: bool

iterdir() → Iterable[S3_Bucket_Key]: Return the S3_Bucket_Key objects contained in the in/under this object.

joinpath(*others) → S3_Bucket_Key | S3_Bucket_Folder

Join multiple path components.

Parameters:: *others¶ – Path components to join.
Returns:: New S3 path object with all components joined.
Return type:: S3_Bucket_Key | S3_Bucket_Folder

json(*, include: set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None = None, exclude: set[int] | set[str] | Mapping[int, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | Mapping[str, set[int] | set[str] | Mapping[int, IncEx | bool] | Mapping[str, IncEx | bool] | bool] | None = None, by_alias: bool = False, exclude_unset: bool = False, exclude_defaults: bool = False, exclude_none: bool = False, encoder: Callable[[Any], Any] | None = PydanticUndefined, models_as_dict: bool = PydanticUndefined, **dumps_kwargs: Any) → str

keepass: If the password_source is KEEPASS, then load a sub-section with the config_wrangler.config_templates.keepass_config.KeepassConfig) settings

keepass_config: If the password_source is KEEPASS, then which root level config item contains the settings for Keepass (must be an instance of config_wrangler.config_templates.keepass_config.KeepassConfig)

keepass_group

If the password_source is KEEPASS, which group in the Keepass database should be searched for an entry with a matching entry.

If is None, then the KeepassConfig.default_group value will be checked. If that is also None, then a ValueError will be raised.

keepass_title: If the password_source is KEEPASS, this is an optional filter on the title of the keepass entries in the group.

key: The S3 object key (path within bucket)

key_exists(key: str | PurePosixPath) → bool

Check if an S3 object exists at the specified key.

Deprecated since version Use: exists() instead.

Parameters:: key¶ (str | PurePosixPath) – S3 key to check.
Returns:: True if the object exists, False otherwise.
Return type:: bool

keyring_section

If the password_source is KEYRING, then which section (AKA system) should this module look for the password in.

See https://pypi.org/project/keyring/ or https://github.com/jaraco/keyring

list_object_keys(key: str | PurePosixPath | None = None, treat_as_folder: bool | None = None) → List[str]

List S3 object keys under a key prefix.

Parameters:

key¶ (str | PurePosixPath | None, optional) – S3 key prefix. If None, uses the instance key.
treat_as_folder¶ (bool | None, optional) – Whether to treat the key as a folder. If None, uses instance setting.

Returns:

List of S3 object keys.

Return type:

List[str]

list_object_paths(key: str | PurePosixPath | None = None, treat_as_folder: bool | None = None) → List[PurePosixPath]

Return the relative paths of objects contained in/under this object.

Parameters:

key¶ (str | PurePosixPath | None, optional) – S3 key prefix. If None, uses the instance key.
treat_as_folder¶ (bool | None, optional) – Whether to treat the key as a folder. If None, uses instance setting.

Returns:

List of relative paths (PurePosixPath) of objects under the key prefix.

Return type:

List[PurePosixPath]

list_objects(key: str | PurePosixPath | None = None, treat_as_folder: bool | None = None) → BucketObjectsCollection

List S3 objects under a key prefix.

Parameters:

key¶ (str | PurePosixPath | None, optional) – S3 key prefix. If None, uses the instance key.
treat_as_folder¶ (bool | None, optional) – Whether to treat the key as a folder by appending ‘/’. If None, uses instance setting.

Returns:

Collection of S3 ObjectSummary objects matching the prefix.

Return type:

BucketObjectsCollection

Raises:

S3ClientError – If the bucket or key doesn’t exist, or other errors occur.

load_into_table_from_s3(database: DatabaseMetadata, table_name: str = None, filename: str = None, delimiter: str = None, region: str = None)[source]

model_computed_fields = {}

model_config = {'validate_assignment': True, 'validate_credentials': True, 'validate_default': True}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

classmethod model_construct(_fields_set: set[str] | None = None, **values: Any) → Self

Creates a new instance of the Model class with validated data.

Creates a new model setting __dict__ and __pydantic_fields_set__ from trusted or pre-validated data. Default values are respected, but no other validation is performed.

!!! note: model_construct() generally respects the model_config.extra setting on the provided model. That is, if model_config.extra == ‘allow’, then all extra passed values are added to the model instance’s __dict__ and __pydantic_extra__ fields. If model_config.extra == ‘ignore’ (the default), then all extra passed values are ignored. Because no validation is performed with a call to model_construct(), having model_config.extra == ‘forbid’ does not result in an error if extra values are passed, but they will be ignored.

Parameters:

_fields_set¶ – A set of field names that were originally explicitly set during instantiation. If provided, this is directly used for the [model_fields_set][pydantic.BaseModel.model_fields_set] attribute. Otherwise, the field names from the values argument will be used.
values¶ – Trusted or pre-validated data dictionary.

Returns:

A new instance of the Model class with validated data.

model_copy(*, update: Mapping[str, Any] | None = None, deep: bool = False) → Self

!!! abstract “Usage Documentation”: [model_copy](../concepts/models.md#model-copy)

Returns a copy of the model.

!!! note: The underlying instance’s [__dict__][object.__dict__] attribute is copied. This might have unexpected side effects if you store anything in it, on top of the model fields (e.g. the value of [cached properties][functools.cached_property]).

Parameters:

update¶ – A mapping of values to update the copied model. Updates are not applied recursively, and no validation is performed on updated values. Only the known fields are updated (if unknown keys are being passed and the model has [extra][pydantic.ConfigDict.extra] set to ‘allow’, they are added as extra data).
deep¶ – Whether a [deep copy][copy.deepcopy] of the model should be performed.

Returns:

New model instance.

!!! abstract “Usage Documentation”: [model_dump](../concepts/serialization.md#python-mode)

Generate a dictionary representation of the model, optionally specifying which fields to include or exclude.

Parameters:

mode¶ – The mode in which to_python should run. If mode is ‘json’, the output will only contain JSON serializable types. If mode is ‘python’, the output may contain non-JSON-serializable Python objects.
include¶ – A set of fields to include in the output.
exclude¶ – A set of fields to exclude from the output.
context¶ – Additional context to pass to the serializer.
by_alias¶ – Whether to use the field’s alias in the dictionary key if defined.
exclude_unset¶ – Whether to exclude fields that have not been explicitly set.
exclude_defaults¶ – Whether to exclude fields that are set to their default value.
exclude_none¶ – Whether to exclude fields that have a value of None.
exclude_computed_fields¶ – Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.
round_trip¶ – If True, dumped values should be valid as input for non-idempotent types such as Json[T].
warnings¶ – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].
fallback¶ – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.
serialize_as_any¶ – Whether to serialize fields with duck-typing serialization behavior.
polymorphic_serialization¶ – Whether to use model and dataclass polymorphic serialization for this call.

Returns:

A dictionary representation of the model.

!!! abstract “Usage Documentation”: [model_dump_json](../concepts/serialization.md#json-mode)

Generates a JSON representation of the model using Pydantic’s to_json method.

Parameters:

indent¶ – Indentation to use in the JSON output. If None is passed, the output will be compact.
ensure_ascii¶ – If True, the output is guaranteed to have all incoming non-ASCII characters escaped. If False (the default), these characters will be output as-is.
include¶ – Field(s) to include in the JSON output.
exclude¶ – Field(s) to exclude from the JSON output.
context¶ – Additional context to pass to the serializer.
by_alias¶ – Whether to serialize using field aliases.
exclude_unset¶ – Whether to exclude fields that have not been explicitly set.
exclude_defaults¶ – Whether to exclude fields that are set to their default value.
exclude_none¶ – Whether to exclude fields that have a value of None.
exclude_computed_fields¶ – Whether to exclude computed fields. While this can be useful for round-tripping, it is usually recommended to use the dedicated round_trip parameter instead.
round_trip¶ – If True, dumped values should be valid as input for non-idempotent types such as Json[T].
warnings¶ – How to handle serialization errors. False/”none” ignores them, True/”warn” logs errors, “error” raises a [PydanticSerializationError][pydantic_core.PydanticSerializationError].
fallback¶ – A function to call when an unknown value is encountered. If not provided, a [PydanticSerializationError][pydantic_core.PydanticSerializationError] error is raised.
serialize_as_any¶ – Whether to serialize fields with duck-typing serialization behavior.
polymorphic_serialization¶ – Whether to use model and dataclass polymorphic serialization for this call.

Returns:

A JSON string representation of the model.

model_dump_non_private(*, mode: Literal['json', 'python'] | str = 'python', exclude: Set[str] | None = None) → dict[str, Any]

property model_extra: Dict[str, Any] | None

Get extra fields set during validation.

Returns:: A dictionary of extra fields, or None if config.extra is not set to “allow”.

model_fields = {'bucket_name': FieldInfo(annotation=str, required=True), 'iam_role': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'keepass': FieldInfo(annotation=Union[KeepassConfig, NoneType], required=False, default=None), 'keepass_config': FieldInfo(annotation=Union[str, NoneType], required=False, default='keepass'), 'keepass_group': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'keepass_title': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'key': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'keyring_section': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'password_source': FieldInfo(annotation=Union[Annotated[PasswordSource, _check_password_source], NoneType], required=False, default=None), 'raw_password': FieldInfo(annotation=Union[str, NoneType], required=False, default=None, repr=False), 'region_name': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'treat_as_folder': FieldInfo(annotation=bool, required=False, default=False), 'user_id': FieldInfo(annotation=Union[str, NoneType], required=False, default=None), 'validate_password_on_load': FieldInfo(annotation=bool, required=False, default=True)}

property model_fields_set: set[str]

Returns the set of fields that have been explicitly set on this model instance.

Returns:

A set of strings representing the fields that have been set,: i.e. that were not filled from defaults.

classmethod model_json_schema(by_alias: bool = True, ref_template: str = '#/$defs/{model}', schema_generator: type[GenerateJsonSchema] = <class 'pydantic.json_schema.GenerateJsonSchema'>, mode: Literal['validation', 'serialization']='validation', *, union_format: Literal['any_of', 'primitive_type_array']='any_of') → Dict[str, Any]

Generates a JSON schema for a model class.

Parameters:

by_alias¶ – Whether to use attribute aliases or not.
ref_template¶ – The reference template.
union_format¶ –
The format to use when combining schemas from unions together. Can be one of:
- ’any_of’: Use the [anyOf](https://json-schema.org/understanding-json-schema/reference/combining#anyOf)
keyword to combine schemas (the default). - ‘primitive_type_array’: Use the [type](https://json-schema.org/understanding-json-schema/reference/type) keyword as an array of strings, containing each type of the combination. If any of the schemas is not a primitive type (string, boolean, null, integer or number) or contains constraints/metadata, falls back to any_of.
schema_generator¶ – To override the logic used to generate the JSON schema, as a subclass of GenerateJsonSchema with your desired modifications
mode¶ – The mode in which to generate the schema.

Returns:

The JSON schema for the given model class.

classmethod model_parametrized_name(params: tuple[TypeForm[Any], ...]) → str

Compute the class name for parametrizations of generic classes.

This method can be overridden to achieve a custom naming scheme for generic BaseModels.

Parameters:: params¶ – Tuple of types of the class. Given a generic class Model with 2 type variables and a concrete model Model[str, int], the value (str, int) would be passed to params.
Returns:: String representing the new class where params are passed to cls as type variables.
Raises:: TypeError – Raised when trying to generate concrete names for non-generic models.

model_post_init(context: Any, /) → None

This function is meant to behave like a BaseModel method to initialize private attributes.

It takes context as an argument since that’s what pydantic-core passes when calling it.

Parameters:

self¶ – The BaseModel instance.
context¶ – The context.

classmethod model_rebuild(*, force: bool = False, raise_errors: bool = True, _parent_namespace_depth: int = 2, _types_namespace: MappingNamespace | None = None) → bool | None

Try to rebuild the pydantic-core schema for the model.

This may be necessary when one of the annotations is a ForwardRef which could not be resolved during the initial attempt to build the schema, and automatic rebuilding fails.

Parameters:

force¶ – Whether to force the rebuilding of the model schema, defaults to False.
raise_errors¶ – Whether to raise errors, defaults to True.
_parent_namespace_depth¶ – The depth level of the parent namespace, defaults to 2.
_types_namespace¶ – The types namespace, defaults to None.

Returns:

Returns None if the schema is already “complete” and rebuilding was not required. If rebuilding _was_ required, returns True if rebuilding was successful, otherwise False.

Validate a pydantic model instance.

If validation fails, the resulting [ValidationError][pydantic_core.ValidationError] reports the rejected locations and values. [Logfire](../integrations/logfire.md) can retain the complete validation input and surrounding trace context for production debugging (see [Troubleshooting validation errors](../errors/troubleshooting.md)).

Parameters:

obj¶ – The object to validate.
strict¶ – Whether to enforce types strictly.
extra¶ – Whether to ignore, allow, or forbid extra data during model validation. See the [extra configuration value][pydantic.ConfigDict.extra] for details.
from_attributes¶ – Whether to extract data from object attributes.
context¶ – Additional context to pass to the validator.
by_alias¶ – Whether to use the field’s alias when validating against the provided input data.
by_name¶ – Whether to use the field’s name when validating against the provided input data.

Raises:

ValidationError – If the object could not be validated.

Returns:

The validated model instance.

!!! abstract “Usage Documentation”: [JSON Parsing](../concepts/json.md#json-parsing)

Validate the given JSON data against the Pydantic model.

A [ValidationError][pydantic_core.ValidationError] raised here reports the rejected locations and values, but may not retain the complete source document. Recording validations with [Logfire](../integrations/logfire.md) keeps the complete JSON input alongside the error — see [Troubleshooting validation errors](../errors/troubleshooting.md).

Parameters:

json_data¶ – The JSON data to validate.
strict¶ – Whether to enforce types strictly.
extra¶ – Whether to ignore, allow, or forbid extra data during model validation. See the [extra configuration value][pydantic.ConfigDict.extra] for details.
context¶ – Extra variables to pass to the validator.
by_alias¶ – Whether to use the field’s alias when validating against the provided input data.
by_name¶ – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

Raises:

ValidationError – If json_data is not a JSON string or the object could not be validated.

Validate the given object with string data against the Pydantic model.

Parameters:

obj¶ – The object containing string data to validate.
strict¶ – Whether to enforce types strictly.
extra¶ – Whether to ignore, allow, or forbid extra data during model validation. See the [extra configuration value][pydantic.ConfigDict.extra] for details.
context¶ – Extra variables to pass to the validator.
by_alias¶ – Whether to use the field’s alias when validating against the provided input data.
by_name¶ – Whether to use the field’s name when validating against the provided input data.

Returns:

The validated Pydantic model.

property name

Get the final component of the S3 key.

Returns:: The file name or final path component.
Return type:: str

nav_to_bucket(bucket_name) → S3_Bucket

nav_to_s3_link(s3_uri: str) → S3_Bucket_Key

open(mode: str = 'r', encoding: str | None = None, errors: str | None = None) → IOBase

Open the S3 object as a file-like object for reading.

Parameters:

mode¶ (str, optional) – File mode (‘r’ for text, ‘rb’ for binary). Default is ‘r’.
encoding¶ (str | None, optional) – Text encoding (for text mode only).
errors¶ (str | None, optional) – Error handling strategy for encoding (for text mode only).

Returns:

File-like object (TextIOWrapper for text mode, BufferedReader for binary).

Return type:

io.IOBase

Raises:

ValueError – If mode is empty or doesn’t start with ‘r’ (only read mode is supported).

property parent

Get the parent S3_Bucket_Key.

Returns:: S3_Bucket_Key representing the parent key.
Return type:: S3_Bucket_Key

property parents

Get all parent S3_Bucket_Key instances.

Returns:: List of S3_Bucket_Key instances representing all parent keys.
Return type:: List[S3_Bucket_Key]

classmethod parse_file(path: str | Path, *, content_type: str | None = None, encoding: str = 'utf8', proto: DeprecatedParseProtocol | None = None, allow_pickle: bool = False) → Self

classmethod parse_obj(obj: Any) → Self

classmethod parse_raw(b: str | bytes, *, content_type: str | None = None, encoding: str = 'utf8', proto: DeprecatedParseProtocol | None = None, allow_pickle: bool = False) → Self

property parts

Get the parts of the S3 path (bucket name + key parts).

Returns:: List containing bucket name followed by key parts.
Return type:: List[str]

password_source: The source to use when getting a password for the user. See PasswordSource for valid values.

raw_password: This is only used for the extremely non-secure CONFIG_FILE password source. The password is stored directly in the config file next to the user_id with the setting name raw_password

read_bytes() → bytes

Read the S3 object contents as bytes.

Returns:: The complete file contents as bytes.
Return type:: bytes

read_text(encoding='utf-8', errors=None) → str

Read the S3 object contents as text.

Parameters:

encoding¶ (str, optional) – Text encoding. Default is ‘utf-8’.
errors¶ (str | None, optional) – Error handling strategy for encoding.

Returns:

The complete file contents as a string.

Return type:

str

region_name

rename(target: str)

Rename (move) this S3 object to a new key.

Parameters:: target¶ (str) – Target S3 key or URI.

Notes

This copies the object to the new location then deletes the original.

property resource: S3ServiceResource

Get the boto3 S3 service resource.

Returns:: The boto3 S3 service resource.
Return type:: S3ServiceResource

scan_files_from_s3(key_prefixes: List[str], local_folder: str | None = None) → List[File_Entry][source]

classmethod schema(by_alias: bool = True, ref_template: str = '#/$defs/{model}') → Dict[str, Any]

classmethod schema_json(*, by_alias: bool = True, ref_template: str = '#/$defs/{model}', **dumps_kwargs: Any) → str

property session: Session

set_as_child(name: str, other_config_item: ConfigHierarchy)

set_session(session: Session)

static split_s3_uri(s3_uri: str) → Tuple[str, str]

property stem

Get the file name without extension.

Returns:: The file name without the final extension.
Return type:: str

property suffix

Get the file extension of the S3 key.

Returns:: The file extension (e.g., ‘.txt’, ‘.json’).
Return type:: str

property suffixes

Get all file extensions of the S3 key.

Returns:: List of all file extensions (e.g., [‘.tar’, ‘.gz’]).
Return type:: List[str]

static translate_config_data(config_data: MutableMapping): Children classes can provide translation logic to allow older config files to be used with newer config class definitions.

treat_as_folder: Whether to treat the key as a folder prefix

unlink(key: str | PurePosixPath | None = None, version_id: str | None = None, missing_ok=False)

Delete an S3 object (pathlib-style interface).

Parameters:

key¶ (str | PurePosixPath | None, optional) – S3 key to delete. If None, uses the instance key.
version_id¶ (str | None, optional) – Specific version ID to delete. If None, deletes the latest version.
missing_ok¶ (bool, optional) – If True, don’t raise an error if the object doesn’t exist. Default is False.

Raises:

ClientError – If the object doesn’t exist and missing_ok is False, or other errors occur.

unload_data(database: DatabaseMetadata, file_format: FileFormat = None, query: str = None, out_folder: str = None, filename: str = None, delimiter: str = None)[source]

classmethod update_forward_refs(**localns: Any) → None

Upload a local file to S3.

Parameters:

local_filename¶ (str | Path) – The local file to upload.
key¶ (str | PurePosixPath | None, optional) – S3 key to upload to. If None, uses the instance key.
extra_args¶ (dict | None, optional) – Extra arguments to pass to boto3 upload_file (e.g., metadata, ACL).
transfer_config¶ (TransferConfig | None, optional) – Transfer configuration for the upload. Defaults to 5GB multipart threshold.
overwrite_mode¶ (OverwriteModes, optional) – Controls overwrite behavior. Default is ALWAYS_OVERWRITE.

upload_file_to_s3(file_key: str, local_file_path: str)[source]

user_id: The user ID to use

classmethod validate(value: Any) → Self

validate_password_on_load: Should config_wrangler query the password source for this password at load (startup) time? If so, it will raise an error if the password is None or an empty string. It does not actually connect or authenticate the user_id & password combination.

with_name(name: str)

Return a new S3_Bucket_Key with the file name changed.

Parameters:: name¶ (str) – The new file name.
Returns:: New S3_Bucket_Key with the modified name.
Return type:: S3_Bucket_Key

with_stem(stem: str)

Return a new S3_Bucket_Key with the file stem changed.

Parameters:: stem¶ (str) – The new file stem (name without extension).
Returns:: New S3_Bucket_Key with the modified stem.
Return type:: S3_Bucket_Key

with_suffix(suffix: str)

Return a new S3_Bucket_Key with the file suffix changed.

Parameters:: suffix¶ (str) – The new file extension (e.g., ‘.txt’).
Returns:: New S3_Bucket_Key with the modified suffix.
Return type:: S3_Bucket_Key

class bi_etl.boto3_helper.s3.FileFormat(*values)[source]

Bases: Enum

csv = 'C'

parquet = 'P'

class bi_etl.boto3_helper.s3.File_Entry(content_path: str, full_local_path: str, source_path: str = None, s3_last_modified: datetime = None, s3_file_size: int = None, bucket_object=None, local_last_modified: datetime = None, local_file_size: int = None)[source]

Bases: object

__init__(content_path: str, full_local_path: str, source_path: str = None, s3_last_modified: datetime = None, s3_file_size: int = None, bucket_object=None, local_last_modified: datetime = None, local_file_size: int = None)[source]

basename()[source]

full_path(base_path)[source]