algorithmia-python/Algorithmia/client.py at develop · algorithmiaio/algorithmia-python · GitHub

History

421 lines (355 loc) · 15.6 KB

Raw

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

138

139

140

141

142

143

144

145

146

147

148

149

150

151

152

153

154

155

156

157

158

159

160

161

162

163

164

165

166

167

168

169

170

171

172

173

174

175

176

177

178

179

180

181

182

183

184

185

186

187

188

189

190

191

192

193

194

195

196

197

198

199

200

201

202

203

204

205

206

207

208

209

210

211

212

213

214

215

216

217

218

219

220

221

222

223

224

225

226

227

228

229

230

231

232

233

234

235

236

237

238

239

240

241

242

243

244

245

246

247

248

249

250

251

252

253

254

255

256

257

258

259

260

261

262

263

264

265

266

267

268

269

270

271

272

273

274

275

276

277

278

279

280

281

282

283

284

285

286

287

288

289

290

291

292

293

294

295

296

297

298

299

300

301

302

303

304

305

306

307

308

309

310

311

312

313

314

315

316

317

318

319

320

321

322

323

324

325

326

327

328

329

330

331

332

333

334

335

336

337

338

339

340

341

342

343

344

345

346

347

348

349

350

351

352

353

354

355

356

357

358

359

360

361

362

363

364

365

366

367

368

369

370

371

372

373

374

375

376

377

378

379

380

381

382

383

384

385

386

387

388

389

390

391

392

393

394

395

396

397

398

399

400

401

402

403

404

405

406

407

408

409

410

411

412

413

414

415

416

417

418

419

420

421

'Algorithmia API Client (python)'

import Algorithmia

from Algorithmia.insights import Insights

from Algorithmia.errors import raiseAlgoApiError

from Algorithmia.algorithm import Algorithm

from Algorithmia.datafile import DataFile, LocalDataFile, AdvancedDataFile

from Algorithmia.datadirectory import DataDirectory, LocalDataDirectory, AdvancedDataDirectory

from algorithmia_api_client import Configuration, DefaultApi, ApiClient

from Algorithmia.util import md5_for_file, md5_for_str

from tempfile import mkstemp

import atexit

import json, re, requests, six, certifi

import tarfile

import os

from time import time

class Client(object):

'Algorithmia Common Library'

handle, ca_cert = None, None

apiKey = None

apiAddress = None

requestSession = None

bearerToken = None

def __init__(self, apiKey=None, apiAddress=None, caCert=None, bearerToken=None):

# Override apiKey with environment variable

config = None

self.requestSession = requests.Session()

if apiKey is None and 'ALGORITHMIA_API_KEY' in os.environ:

apiKey = os.environ['ALGORITHMIA_API_KEY']

elif bearerToken is None and 'ALGORITHMIA_BEARER_TOKEN' in os.environ:

bearerToken = os.environ['ALGORITHMIA_BEARER_TOKEN']

self.bearerToken = bearerToken

self.apiKey = apiKey

if apiAddress is not None:

self.apiAddress = apiAddress

else:

self.apiAddress = Algorithmia.getApiAddress()

if caCert == False:

self.requestSession.verify = False

self.requestSession.trust_env = False

config = Configuration(use_ssl=False)

elif caCert is None and 'REQUESTS_CA_BUNDLE' in os.environ:

caCert = os.environ.get('REQUESTS_CA_BUNDLE')

self.catCerts(caCert)

self.requestSession.verify = self.ca_cert

elif caCert is not None and 'REQUESTS_CA_BUNDLE' not in os.environ:

self.catCerts(caCert)

self.requestSession.verify = self.ca_cert

elif caCert is not None and 'REQUESTS_CA_BUNDLE' in os.environ:

# if both are available, use the one supplied in the constructor. I assume that a user supplying a cert in initialization wants to use that one.

self.catCerts(caCert)

self.requestSession.verify = self.ca_cert

if not config:

config = Configuration()

config.api_key['Authorization'] = self.apiKey

config.host = "{}/v1".format(self.apiAddress)

self.manageApi = DefaultApi(ApiClient(config))

def algo(self, algoRef):

return Algorithm(self, algoRef)

def username(self):

username = next(self.dir("").list()).path

return username

def scms(self):

url = "/v1/scms"

response = self.getJsonHelper(url)

return response

def file(self, dataUrl, cleanup=False):

if dataUrl.startswith('file://'):

return LocalDataFile(self, dataUrl)

else:

return AdvancedDataFile(self, dataUrl, cleanup)

def dir(self, dataUrl):

if dataUrl.startswith('file://'):

return LocalDataDirectory(self, dataUrl)

else:

return AdvancedDataDirectory(self, dataUrl)

def create_user(self, requestString):

url = "/v1/users"

response = self.postJsonHelper(url, input_object=requestString)

return response

def get_org_types(self):

url = "/v1/organization/types"

response = self.getHelper(url)

return json.loads(response.content.decode("utf-8"))

def create_org(self, requestString):

url = "/v1/organizations"

type = requestString["type_id"]

id, error = self.convert_type_id(type)

requestString["type_id"] = id

response = self.postJsonHelper(url=url, input_object=requestString)

if (error != "") and (response["error"] is not None):

response["error"]["message"] = error

return response

def get_org(self, org_name):

url = "/v1/organizations/" + org_name

response = self.getHelper(url)

return json.loads(response.content.decode("utf-8"))

def edit_org(self, org_name, requestString):

url = "/v1/organizations/" + org_name

type = requestString["type_id"]

id, error = self.convert_type_id(type)

requestString["type_id"] = id

data = json.dumps(requestString).encode('utf-8')

response = self.putHelper(url, data)

if (error != "") and (response["error"] is not None):

response["error"]["message"] = error

return response

def invite_to_org(self, orgname, username):

url = "/v1/organizations/" + orgname + "/members/" + username

response = self.putHelper(url, data={})

return response

def get_template(self, envid, dest, save_tar=False):

url = "/v1/algorithm-environments/edge/environment-specifications/" + envid + "/template"

filename = "template.tar.gz"

if not os.path.exists(dest):

os.makedirs(dest)

filepath = os.path.join(dest, filename)

response = self.getStreamHelper(url)

if response.ok:

with open(filepath, 'wb') as f:

for chunk in response.iter_content(chunk_size=1024 * 8):

if chunk:

f.write(chunk)

f.flush()

os.fsync(f.fileno())

tar = tarfile.open(filepath, "r:gz")

tar.extractall(dest)

tar.close()

if not save_tar:

try:

os.remove(filepath)

except OSError as e:

print(e)

return response

else:

return json.loads(response.content.decode("utf-8"))

def get_environment(self, language):

url = "/v1/algorithm-environments/edge/languages/" + language + "/environments"

response = self.getHelper(url)

return response.json()

def get_supported_languages(self):

url = "/v1/algorithm-environments/edge/languages"

response = self.getHelper(url)

return response.json()

def get_secret_providers(self):

url = "/v1/secret-provider"

api_response = self.getJsonHelper(url)

return api_response

def get_organization_errors(self, org_name):

"""Gets the errors for the organization.

Args:

self (Client): The instance of the Client class.

org_name (str): The identifier for the organization.

Returns:

Any: A JSON-encoded response from the API.

"""

url = '/v1/organizations/%s/errors' % org_name

response = self.getHelper(url)

return response.json()

def get_user_errors(self, user_id):

"""Gets the errors for a specific user.

Args:

self (Client): The instance of the Client class.

user_id (str): The identifier for the user.

Returns:

Any: A JSON-encoded response from the API.

"""

url = '/v1/users/%s/errors' % user_id

response = self.getHelper(url)

return response.json()

def get_algorithm_errors(self, algorithm_id):

"""Gets the errors for a specific algorithm.

Args:

self (Client): The instance of the Client class.

algorithm_id (str): The identifier for the algorithm.

Returns:

Any: A JSON-encoded response from the API.

"""

url = '/v1/algorithms/%s/errors' % algorithm_id

return self.getJsonHelper(url)

# Used to send insight data to Algorithm Queue Reader in cluster

def report_insights(self, insights):

return Insights(insights)

# Used internally to post json to the api and parse json response

def postJsonHelper(self, url, input_object, parse_response_as_json=True, retry=False, **query_parameters):

headers = {}

if self.apiKey is not None:

headers['Authorization'] = self.apiKey

elif self.bearerToken is not None:

headers['Authorization'] = 'Bearer ' + self.bearerToken

input_json = None

if input_object is None:

input_json = json.dumps(None).encode('utf-8')

headers['Content-Type'] = 'application/json'

elif isinstance(input_object, six.string_types):

input_json = input_object.encode('utf-8')

headers['Content-Type'] = 'text/plain'

elif isinstance(input_object, bytearray) or isinstance(input_object, bytes):

input_json = bytes(input_object)

headers['Content-Type'] = 'application/octet-stream'

else:

input_json = json.dumps(input_object).encode('utf-8')

headers['Content-Type'] = 'application/json'

response = self.requestSession.post(self.apiAddress + url, data=input_json, headers=headers,

params=query_parameters)

if 200 <= response.status_code <= 299:

if parse_response_as_json:

response = response.json()

if 'error' in response:

raise raiseAlgoApiError(response)

else:

return response

else:

return response

elif retry:

return self.postJsonHelper(url, input_object, parse_response_as_json, False, **query_parameters)

else:

raise raiseAlgoApiError(response)

# Used internally to http get a file

def getHelper(self, url, **query_parameters):

headers = {}

if self.apiKey is not None:

headers['Authorization'] = self.apiKey

elif self.bearerToken is not None:

headers['Authorization'] = 'Bearer ' + self.bearerToken

return self.requestSession.get(self.apiAddress + url, headers=headers, params=query_parameters)

def getJsonHelper(self, url, **query_parameters):

headers = {}

if self.apiKey is not None:

headers['Authorization'] = self.apiKey

elif self.bearerToken is not None:

headers['Authorization'] = 'Bearer ' + self.bearerToken

response = self.requestSession.get(self.apiAddress + url, headers=headers, params=query_parameters)

if 200 <= response.status_code <= 299:

response = response.json()

if 'error' in response:

raise raiseAlgoApiError(response)

else:

return response

else:

if response.content is not None:

response = response.json()

raise raiseAlgoApiError(response)

def getStreamHelper(self, url, **query_parameters):

headers = {}

if self.apiKey is not None:

headers['Authorization'] = self.apiKey

elif self.bearerToken is not None:

headers['Authorization'] = 'Bearer ' + self.bearerToken

return self.requestSession.get(self.apiAddress + url, headers=headers, params=query_parameters, stream=True)

def patchHelper(self, url, params):

headers = {'content-type': 'application/json'}

if self.apiKey is not None:

headers['Authorization'] = self.apiKey

elif self.bearerToken is not None:

headers['Authorization'] = 'Bearer ' + self.bearerToken

return self.requestSession.patch(self.apiAddress + url, headers=headers, data=json.dumps(params))

# Used internally to get http head result

def headHelper(self, url):

headers = {}

if self.apiKey is not None:

headers['Authorization'] = self.apiKey

elif self.bearerToken is not None:

headers['Authorization'] = 'Bearer ' + self.bearerToken

return self.requestSession.head(self.apiAddress + url, headers=headers)

# Used internally to http put a file

def putHelper(self, url, data):

headers = {}

if self.apiKey is not None:

headers['Authorization'] = self.apiKey

elif self.bearerToken is not None:

headers['Authorization'] = 'Bearer ' + self.bearerToken

if isJson(data):

headers['Content-Type'] = 'application/json'

response = self.requestSession.put(self.apiAddress + url, data=data, headers=headers)

if response._content == b'':

return response

if 200 <= response.status_code <= 299:

response = response.json()

if 'error' in response:

raise raiseAlgoApiError(response)

else:

return response

else:

raise raiseAlgoApiError(response)

# Used internally to http delete a file

def deleteHelper(self, url):

headers = {}

if self.apiKey is not None:

headers['Authorization'] = self.apiKey

elif self.bearerToken is not None:

headers['Authorization'] = 'Bearer ' + self.bearerToken

response = self.requestSession.delete(self.apiAddress + url, headers=headers)

if response.reason == "No Content":

return response

return response.json()

# Used internally to concatonate given custom cert with built in certificate store.

def catCerts(self, customCert):

self.handle, self.ca_cert = mkstemp(suffix=".pem")

# wrapped all in the with context handler to prevent unclosed files

with open(customCert, 'r') as custom_cert, \

open(self.ca_cert, 'w') as ca, \

open(certifi.where(), 'r') as cert:

new_cert = custom_cert.read() + cert.read()

ca.write(new_cert)

atexit.register(self.exit_handler)

# User internally to convert type id name to uuid

def convert_type_id(self, type):

id = ""

error = ""

types = self.get_org_types()

for enumtype in types:

if type == enumtype["name"]:

id = enumtype["id"]

error = ""

break

else:

error = "invalid type_id"

return (id, error)

# Used internally to clean up temporary files

def exit_handler(self):

try:

os.close(self.handle)

os.unlink(self.ca_cert)

except OSError as e:

print(e)

# Used by CI/CD automation for freezing model manifest files, and by the CLI for manual freezing

def freeze(self, manifest_path, manifest_output_dir="."):

if os.path.exists(manifest_path):

with open(manifest_path, 'r') as f:

manifest_file = json.load(f)

manifest_file['timestamp'] = str(time())

required_files = manifest_file['required_files']

optional_files = manifest_file['optional_files']

for i in range(len(required_files)):

uri = required_files[i]['source_uri']

local_file = self.file(uri).getFile(as_path=True)

md5_checksum = md5_for_file(local_file)

required_files[i]['md5_checksum'] = md5_checksum

for i in range(len(optional_files)):

uri = required_files[i]['source_uri']

local_file = self.file(uri).getFile(as_path=True)

md5_checksum = md5_for_file(local_file)

required_files[i]['md5_checksum'] = md5_checksum

lock_md5_checksum = md5_for_str(str(manifest_file))

manifest_file['lock_checksum'] = lock_md5_checksum

with open(manifest_output_dir + '/' + 'model_manifest.json.freeze', 'w') as f:

json.dump(manifest_file, f)

else:

print("Expected to find a model_manifest.json file, none was discovered in working directory")

def isJson(myjson):

try:

json_object = json.loads(myjson)

except (ValueError, TypeError) as e:

return False

return True

algorithmia-python/Algorithmia/client.py at develop · algorithmiaio/algorithmia-python · GitHub

Uh oh!

Expand file tree

Breadcrumbs

client.py

Latest commit

History

Breadcrumbs

client.py

File metadata and controls