Closed bubanoid closed 2 years ago
oooh, yeah this most likely needs to be patched in, will have a look into it
I just added this test which worked fine
@pytest.mark.asyncio
async def test_s3_copy_from(event_loop, s3_client, s3_resource, bucket_name):
data = b'Hello World\n'
await s3_client.create_bucket(Bucket=bucket_name)
fh = BytesIO()
fh.write(data)
fh.seek(0)
await s3_client.upload_fileobj(fh, bucket_name, 'test_file')
resource = await s3_resource.Object(bucket_name, "new_test_file")
copy_source = bucket_name + "/test_file"
copy_result = await resource.copy_from(CopySource=copy_source)
assert 'CopyObjectResult' in copy_result
resp = await s3_client.get_object(Bucket=bucket_name, Key='new_test_file')
assert (await resp['Body'].read()) == data
I suspect its unhappy as both resource.Object()
AND .copy_from
each return a coroutine, so you'll want to await resource.Object into a variable then await variable.copy_from
Yeah, you are right. Intermediate variable res = await resource.Object(BUCKET, PATH_NEW.format(i))
solves the problem.
Thank you!
details:
details:
details:
Description
There is no
copy_from
method for theObject
object ofs3
resourceWhat I Did
raises an error:
however the synchronous version of this script works without errors.
It would be great if you add this method to the library.