forked from cloudconvert/cloudconvert-python
-
Notifications
You must be signed in to change notification settings - Fork 0
Expand file tree
/
Copy pathprocess.py
More file actions
137 lines (98 loc) · 4.16 KB
/
Copy pathprocess.py
File metadata and controls
137 lines (98 loc) · 4.16 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
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
109
110
111
112
113
114
115
116
117
118
119
120
121
122
123
124
125
126
127
128
129
130
131
132
133
134
135
136
137
import os
import shutil
import time
from .exceptions import (
APIError, HTTPError, BadRequest, ConversionFailed, TemporaryUnavailable, InvalidResponse, InvalidParameterException
)
class Process(object):
"""
Process Object wrapper CloudConvert API
"""
data = {}
def __init__(self, api, url=None):
"""
Creates a new Process instance
:param Api api: API Instance
:param str url: The Process URL
"""
self.api = api
self.url = url
def refresh(self, parameters = None):
"""
Refresh process data from API
:param parameters: Parameters for creating the Process. See https://cloudconvert.com/apidoc#start
:raises APIError: if the CloudConvert API returns an error
"""
self.data = self.api.get(self.url, parameters)
return self
def start(self, parameters):
"""
Starts the Process
:param parameters: Parameters for creating the Process. See https://cloudconvert.com/apidoc#start
:raises APIError: if the CloudConvert API returns an error
"""
self.data = self.api.post(self.url, parameters)
return self
def delete(self):
"""
Delete process from API
:raises APIError: if the CloudConvert API returns an error
"""
self.api.delete(self.url)
return self
def wait(self, interval = 1):
"""
Waits for the Process to finish (or end with an error). Checks the conversion status every interval seconds.
:param int interval: Interval in seconds
:raises APIError: if the CloudConvert API returns an error
"""
while self['step']!='finished' and self['step'] !='error':
self.refresh()
time.sleep(interval)
return self
def download(self, localfile = None, remotefile = None):
"""
Download process file from API
:param str localfile: Local file name (or directory) the file should be downloaded to
:param str remotefile: Remote file name which should be downloaded (if there are multiple output files available)
:raises APIError: if the CloudConvert API returns an error
"""
if localfile is not None and os.path.isdir(localfile) and 'filename' in self.data.get('output', {}):
## localfile is directory
localfile = os.path.normpath(localfile) + os.sep + (remotefile if remotefile is not None else self['output']['filename'])
elif localfile is None and 'filename' in self.data.get('output', {}):
## localfile is not set -> set it to output filename
localfile = remotefile if remotefile is not None else self['output']['filename']
if localfile is None or os.path.isdir(localfile):
raise InvalidParameterException("localfile parameter is not set correctly")
if 'url' not in self.data.get('output', {}):
raise APIError("There is no output file available (yet)")
r = self.api.rawCall("GET", self['output']['url'] + ("/" + remotefile if remotefile else ""), stream=True)
with open(localfile, 'wb') as f:
r.raw.decode_content = True
shutil.copyfileobj(r.raw, f)
return self
def downloadAll(self, directory = None):
"""
Download all output process files from API
:param str directory: Local directory the files should be downloaded to
:raises APIError: if the CloudConvert API returns an error
"""
if 'files' not in self.data.get('output', {}):
## there are not multiple output files -> do normal download
return self.download(localfile=directory)
for file in self["output"]["files"]:
self.download(localfile=directory, remotefile=file)
return self
def __getitem__(self, item):
"""
Make process status from API available as object attributes.
Examples:
process['step']
process['message']
"""
if self.data.get(item):
return self.data.get(item)
else:
# Default behaviour
raise AttributeError