Coverage for src/io_collection/keys/copy_key.py: 100%

1import shutil 

2from pathlib import Path 

3 

4import boto3 

5 

6 

7def copy_key(location: str, old_key: str, new_key: str) -> None: 

8 """ 

9 Copy object key at specified location. 

10 

11 Method will copy the object in an S3 bucket if the location begins with the 

12 **s3://** protocol, otherwise it assumes the location is a local path. 

13 

14 Parameters 

15 ---------- 

16 location 

17 Object location (local path or S3 bucket). 

18 old_key 

19 Old object key. 

20 new_key 

21 New object key. 

22 """ 

23 

24 if location[:5] == "s3://": 

25 _copy_key_on_s3(location[5:], old_key, new_key) 

26 else: 

27 _copy_key_on_fs(location, old_key, new_key) 

28 

29 

30def _copy_key_on_fs(path: str, old_key: str, new_key: str) -> None: 

31 """ 

32 Copy object key on local file system. 

33 

34 Parameters 

35 ---------- 

36 path 

37 Local object path. 

38 old_key 

39 Old object key. 

40 new_key 

41 New object key. 

42 """ 

43 

44 full_old_path = Path(path) / old_key 

45 full_new_path = Path(path) / new_key 

46 full_new_path.parent.mkdir(parents=True, exist_ok=True) 

47 shutil.copyfile(full_old_path, full_new_path) 

48 

49 

50def _copy_key_on_s3(bucket: str, old_key: str, new_key: str) -> None: 

51 """ 

52 Copy object key in AWS S3 bucket. 

53 

54 Parameters 

55 ---------- 

56 bucket 

57 AWS S3 bucket name. 

58 old_key 

59 Old object key. 

60 new_key 

61 New object key. 

62 """ 

63 

64 s3_client = boto3.client("s3") 

65 s3_client.copy_object(Bucket=bucket, CopySource=f"{bucket}/{old_key}", Key=new_key)