[Python] Pydantic CLIApp/CLISubCommand Not Loading Parameters with Env Vars

I am currently building a CLI using Pydantic. One of the sub commands has 2 parameters I would like to load via an Env variable. I was able to load the environment variables on its own when I instantiated the class directly but now that it is a CliSubCommand. I am having issues loading the two parameters via Env variables. I also was able to get the CLI working when AWSStsGcp class was inheriting from BaseModel by providing via the command line. Pydantic is throwing an error saying the two AWS flags are required. They are located in the aws_sts_gcp.py model -> aws_access_key and aws_secret_key.

I also don't want it to be a mandatory global config, since it is currently only relevant for one command.

root_tool.py


from pydantic import BaseModel
from pydantic_settings import CliSubCommand, CliApp
from python_tools.gcp.models.aws_sts_gcp import AwsStsGcp

class DummyCommand(BaseModel):
project_id: str

def cli_cmd(self) -> None:
print(f'This is a dummy command.{self.project_id}"')

class Tool(BaseModel):
aws_sts: CliSubCommand[AwsStsGcp]
dummy: CliSubCommand[DummyCommand]

def cli_cmd(self) -> None:
CliApp.run_subcommand(self)


aws_sts_gcp.py

from pydantic import SecretStr
from pydantic_settings import BaseSettings, SettingsConfigDict

from python_tools.gcp.aws_gcs_transfer import create_aws_transfer_job

class AwsStsGcp(BaseSettings, cli_parse_args=True):
model_config = SettingsConfigDict(env_file='.env', env_file_encoding='utf-8')
destination_bucket: str
src_bucket: str
manifest_path: str | None = None
aws_access_key: SecretStr
aws_secret_key: SecretStr
tranfer_name: str
project_id: str

def cli_cmd(self) -> None:
create_aws_transfer_job(self)



cli_runner.py

from python_tools.gcp.models.root_tool import Tool
from pydantic_settings import CliApp

CliApp.run(Tool)


aws_gcs_transfer.py

from google.cloud.storage_transfer_v1 import (
StorageTransferServiceClient,
TransferJob,
TransferSpec,
TransferManifest,
AwsS3Data,
AwsAccessKey,
GcsData,
RunTransferJobRequest,
CreateTransferJobRequest
)
#from python_tools.gcp.models.aws_sts_gcp import AwsStsGcp
from python_tools.consts import timestr
from python_tools.logging import logger

import time

def create_aws_transfer_job(transfer_details) -> None:
s3_config = None

transfer_manifest = None

client = StorageTransferServiceClient()
s3_config = AwsS3Data(
bucket_name=transfer_details.src_bucket,
aws_access_key=AwsAccessKey(
access_key_id=transfer_details.aws_access_key.get_secret_value(), secret_access_key=transfer_details.aws_secret_key.get_secret_value()
)
)

gcs_dest = GcsData(bucket_name=transfer_details.destination_bucket)

if transfer_details.manifest_path is not None:
transfer_manifest = TransferManifest(location=transfer_details.manifest_path)

sts_spec = TransferSpec(gcs_data_sink=gcs_dest, aws_s3_data_source=s3_config, transfer_manifest=transfer_manifest)
timestamp = time.strftime(timestr)
name = f"transferJobs/{transfer_details.tranfer_name}-{timestamp}"
description = "Automated STS Job created from Python Tools."
sts_job = TransferJob(
project_id=transfer_details.project_id,
name=name,
description=description,
transfer_spec=sts_spec,
status=TransferJob.Status.ENABLED,
)
job_request = CreateTransferJobRequest(transfer_job=sts_job)

logger.info(f"Starting Transfer Job for Job ID: {name}")
transfer_request = RunTransferJobRequest(project_id=transfer_details.project_id, job_name=name)

client.create_transfer_job(request=job_request)
client.run_transfer_job(request=transfer_request)


.env

AWS_ACCESS_KEY = "test"
AWS_SECRET_KEY = "test"

Continue reading...