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

1from pathlib import Path 

2 

3import boto3 

4 

5 

6def change_key(location: str, old_key: str, new_key: str) -> None: 

7 """ 

8 Change object key at specified location. 

9 

10 Method will change the object in an S3 bucket if the location begins with 

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

12 

13 Parameters 

14 ---------- 

15 location 

16 Object location (local path or S3 bucket). 

17 old_key 

18 Old object key. 

19 new_key 

20 New object key. 

21 """ 

22 

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

24 _change_key_on_s3(location[5:], old_key, new_key) 

25 else: 

26 _change_key_on_fs(location, old_key, new_key) 

27 

28 

29def _change_key_on_fs(path: str, old_key: str, new_key: str) -> None: 

30 """ 

31 Change object key on local file system. 

32 

33 Parameters 

34 ---------- 

35 path 

36 Local object path. 

37 old_key 

38 Old object key. 

39 new_key 

40 New object key. 

41 """ 

42 

43 full_old_path = Path(path) / old_key 

44 full_new_path = Path(path) / new_key 

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

46 full_old_path.rename(full_new_path) 

47 

48 

49def _change_key_on_s3(bucket: str, old_key: str, new_key: str) -> None: 

50 """ 

51 Change object key in AWS S3 bucket. 

52 

53 Parameters 

54 ---------- 

55 bucket 

56 AWS S3 bucket name. 

57 old_key 

58 Old object key. 

59 new_key 

60 New object key. 

61 """ 

62 

63 s3_client = boto3.client("s3") 

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

65 s3_client.delete_object(Bucket=bucket, Key=old_key)