vodorod/Dorod-Sky
1
0
Fork 0
Code Issues Pull Requests Actions 1 Packages Projects Releases Wiki Activity

basic test for s3 artifact upload logic (#2681)

Browse Source
This commit is contained in:
Asher Foa
2025-06-11 14:23:58 -04:00
committed by GitHub
parent ef19c0265e
commit c1e19d27d3
5 changed files with 628 additions and 63 deletions
Download Patch File Download Diff File Expand all files Collapse all files

10
skyvern/forge/sdk/api/aws.py
View File

@@ -37,21 +37,25 @@ class AsyncAWSClient:
aws_access_key_id: str | None = None,
aws_secret_access_key: str | None = None,
region_name: str | None = None,
endpoint_url: str | None = None,
) -> None:
self.region_name = region_name or settings.AWS_REGION
self._endpoint_url = endpoint_url
self.session = aioboto3.Session(
aws_access_key_id=aws_access_key_id,
aws_secret_access_key=aws_secret_access_key,
)
def _ecs_client(self) -> ECSClient:
return self.session.client(AWSClientType.ECS, region_name=self.region_name)
return self.session.client(AWSClientType.ECS, region_name=self.region_name, endpoint_url=self._endpoint_url)
def _secrets_manager_client(self) -> SecretsManagerClient:
return self.session.client(AWSClientType.SECRETS_MANAGER, region_name=self.region_name)
return self.session.client(
AWSClientType.SECRETS_MANAGER, region_name=self.region_name, endpoint_url=self._endpoint_url
)
def _s3_client(self) -> S3Client:
return self.session.client(AWSClientType.S3, region_name=self.region_name)
return self.session.client(AWSClientType.S3, region_name=self.region_name, endpoint_url=self._endpoint_url)
async def get_secret(self, secret_name: str) -> str | None:
try: