Início / coding / Perguntas / 79565427
Accepted
CodyK
Asked: 2025-04-10 06:30:14 +0800 CST

Pydantic CLIApp/CLISubCommand com variáveis ​​de ambiente

  • 772

Atualmente, estou construindo uma CLI usando o Pydantic . Um dos subcomandos tem 2 parâmetros que eu gostaria de carregar por meio de uma variável de ambiente. Consegui carregar as variáveis ​​de ambiente por conta própria quando instanciei a classe diretamente, mas agora que ela é uma variável de ambiente CliSubCommand, estou com problemas para carregar os dois parâmetros por meio de variáveis ​​de ambiente. Também consegui fazer a CLI funcionar quando AWSStsGcpa classe estava herdando de BaseModel, fornecendo-a por meio da linha de comando. O Pydantic está apresentando um erro informando que as duas flags da AWS são necessárias. Elas estão localizadas no modelo aws_sts_gcp.py -> aws_access_keye aws_secret_key.

Também não quero que seja uma configuração global obrigatória, já que atualmente ela só é relevante para um comando.

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"

Também tentei usar BaseSettings na raiz assim.

from pydantic import BaseModel
from pydantic_settings import CliSubCommand, CliApp, BaseSettings, SettingsConfigDict

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(BaseSettings, cli_parse_args=True):
    model_config = SettingsConfigDict(env_file='.env', env_file_encoding='utf-8', extra='ignore')

    aws_sts: CliSubCommand[AwsStsGcp]
    dummy: CliSubCommand[DummyCommand]

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

from pydantic import SecretStr, BaseModel, Field  
from python_tools.gcp.aws_gcs_transfer import create_aws_transfer_job 

class AwsStsGcp(BaseModel):
    destination_bucket: str
    src_bucket: str
    manifest_path: str | None = None
    aws_access_key: SecretStr = Field(alias='AWS_ACCESS_KEY', env="AWS_ACCESS_KEY")
    aws_secret_key: SecretStr = Field(alias='AWS_SECRET_KEY',  env="AWS_SECRET_KEY")
    tranfer_name: str
    project_id: str

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

1 respostas

  1. Best Answer

    Percebi que precisava usar o env_nested_delimiterpara definir variáveis ​​em modelos aninhados. E atualizar o nome da variável Env para

    AWS_STS__AWS_ACCESS_KEY = "test"
AWS_STS__AWS_SECRET_KEY = "test"
    • 0

relate perguntas