-
Notifications
You must be signed in to change notification settings - Fork 8
Expand file tree
/
Copy pathaws_util.py
More file actions
61 lines (51 loc) · 1.9 KB
/
Copy pathaws_util.py
File metadata and controls
61 lines (51 loc) · 1.9 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
# -*- coding: UTF-8 -*-
import subprocess
import sys
from enum import Enum
class AWSCMD(Enum):
Upload = "aws s3 cp {local_path} s3://{bucket_url}/{remote_path}"
Delete = "aws s3 rm s3://{bucket_url}/{remote_path}"
class AWSHelper:
""" 亚马逊 AWS S3 工具类 """
def __init__(self, bucket_url):
""" 初始化
Args:
bucket_url: 桶的链接 s3://bucket_url
"""
self.bucket_url = bucket_url
def __str__(self):
return "aws information >\nbucket_url:%s" % self.bucket_url
def __function(self, _enum, _local_path=None, _remote_path=None, _bucket_url=None):
"""
执行方法
Args:
_enum: 枚举类型
_local_path: 本地文件
_remote_path: 远端文件
_bucket_url: 桶的链接 s3://bucket_url
"""
_command = _enum.value.format(local_path=_local_path, remote_path=_remote_path, _bucket_url=self.bucket_url)
sys.stdout.write(_command + "\n")
sys.stdout.flush()
_process = subprocess.Popen(_command, shell=True)
_process.wait()
if _process.returncode != 0:
sys.stdout.write("{enum_type} failed ! error : {error}\n".format(enum_type=_enum, error=_process.stderr))
else:
sys.stdout.write("{enum_type} successful !\n".format(enum_type=_enum))
sys.stdout.flush()
def upload(self, local_path, remote_path):
""" 上传文件
上传本地文件到远端文件，有会覆盖
Args:
local_path: 本地文件
remote_path: 远端文件
"""
self.__function(AWSCMD.Upload, local_path, remote_path)
def delete(self, remote_path):
""" 删除文件
删除远端文件，没有不会报错
Args:
remote_path: 远端文件
"""
self.__function(AWSCMD.Delete, _remote_path=remote_path)