cm_api.endpoints.types

1 # Licensed to Cloudera, Inc. under one 2 # or more contributor license agreements. See the NOTICE file 3 # distributed with this work for additional information 4 # regarding copyright ownership. Cloudera, Inc. licenses this file 5 # to you under the Apache License, Version 2.0 (the 6 # "License"); you may not use this file except in compliance 7 # with the License. You may obtain a copy of the License at 8 # 9 # http://www.apache.org/licenses/LICENSE-2.0 10 # 11 # Unless required by applicable law or agreed to in writing, software 12 # distributed under the License is distributed on an "AS IS" BASIS, 13 # WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. 14 # See the License for the specific language governing permissions and 15 # limitations under the License. 16 17 try: 18 import json 19 except ImportError: 20 import simplejson as json 21 22 import datetime 23 import time 24 25 __docformat__ = "epytext"

26 27 -class Attr(object):

28 """ 29 Encapsulates information about an attribute in the JSON encoding of the 30 object. It identifies properties of the attribute such as whether it's 31 read-only, its type, etc. 32 """ 33 DATE_FMT = "%Y-%m-%dT%H:%M:%S.%fZ" 34

35 - def __init__(self, atype=None, rw=True, is_api_list=False):

36 self._atype = atype 37 self._is_api_list = is_api_list 38 self.rw = rw

39

40 - def to_json(self, value, preserve_ro):

41 """ 42 Returns the JSON encoding of the given attribute value. 43 44 If the value has a 'to_json_dict' object, that method is called. Otherwise, 45 the following values are returned for each input type: 46 - datetime.datetime: string with the API representation of a date. 47 - dictionary: if 'atype' is ApiConfig, a list of ApiConfig objects. 48 - python list: python list (or ApiList) with JSON encoding of items 49 - the raw value otherwise 50 """ 51 if hasattr(value, 'to_json_dict'): 52 return value.to_json_dict(preserve_ro) 53 elif isinstance(value, dict) and self._atype == ApiConfig: 54 return config_to_api_list(value) 55 elif isinstance(value, datetime.datetime): 56 return value.strftime(self.DATE_FMT) 57 elif isinstance(value, list): 58 if self._is_api_list: 59 return ApiList(value).to_json_dict() 60 else: 61 return [ self.to_json(x, preserve_ro) for x in value ] 62 else: 63 return value

64

65 - def from_json(self, resource_root, data):

66 """ 67 Parses the given JSON value into an appropriate python object. 68 69 This means: 70 - a datetime.datetime if 'atype' is datetime.datetime 71 - a converted config dictionary or config list if 'atype' is ApiConfig 72 - if the attr is an API list, an ApiList with instances of 'atype' 73 - an instance of 'atype' if it has a 'from_json_dict' method 74 - a python list with decoded versions of the member objects if the input 75 is a python list. 76 - the raw value otherwise 77 """ 78 if data is None: 79 return None 80 81 if self._atype == datetime.datetime: 82 return datetime.datetime.strptime(data, self.DATE_FMT) 83 elif self._atype == ApiConfig: 84 # ApiConfig is special. We want a python dictionary for summary views, 85 # but an ApiList for full views. Try to detect each case from the JSON 86 # data. 87 if not data['items']: 88 return { } 89 first = data['items'][0] 90 return json_to_config(data, len(first) == 2) 91 elif self._is_api_list: 92 return ApiList.from_json_dict(self._atype, data, resource_root) 93 elif isinstance(data, list): 94 return [ self.from_json(resource_root, x) for x in data ] 95 elif hasattr(self._atype, 'from_json_dict'): 96 return self._atype.from_json_dict(data, resource_root) 97 else: 98 return data

99

100 -class ROAttr(Attr):

101 """ 102 Subclass that just defines the attribute as read-only. 103 """

104 - def __init__(self, atype=None, is_api_list=False):

105 Attr.__init__(self, atype=atype, rw=False, is_api_list=is_api_list)

106

107 -class BaseApiObject(object):

108 """ 109 The BaseApiObject helps with (de)serialization from/to JSON. 110 111 The derived class has two ways of defining custom attributes: 112 - Overwriting the '_ATTRIBUTES' field with the attribute dictionary 113 - Override the _get_attributes() method, in case static initialization of 114 the above field is not possible. 115 116 It's recommended that the _get_attributes() implementation do caching to 117 avoid computing the dictionary on every invocation. 118 119 The derived class's constructor must call the base class's init() static 120 method. All constructor arguments (aside from self and resource_root) must 121 be keywords arguments with default values (typically None), or 122 from_json_dict() will not work. 123 """ 124 125 _ATTRIBUTES = { } 126 _WHITELIST = ( '_resource_root', '_attributes' ) 127 128 @classmethod

129 - def _get_attributes(cls):

130 """ 131 Returns a map of property names to attr instances (or None for default 132 attribute behavior) describing the properties of the object. 133 134 By default, this method will return the class's _ATTRIBUTES field. 135 Classes can override this method to do custom initialization of the 136 attributes when needed. 137 """ 138 return cls._ATTRIBUTES

139 140 @staticmethod

141 - def init(obj, resource_root, attrs=None):

142 """ 143 Wraper around the real constructor to avoid issues with the 'self' 144 argument. Call like this, from a subclass's constructor: 145 146 BaseApiObject.init(self, locals()) 147 """ 148 # This works around http://bugs.python.org/issue2646 149 # We use unicode strings as keys in kwargs. 150 str_attrs = { } 151 if attrs: 152 for k, v in attrs.iteritems(): 153 if k not in ('self', 'resource_root'): 154 str_attrs[k] = v 155 BaseApiObject.__init__(obj, resource_root, **str_attrs)

156

157 - def __init__(self, resource_root, **attrs):

158 """ 159 Initializes internal state and sets all known writable properties of the 160 object to None. Then initializes the properties given in the provided 161 attributes dictionary. 162 163 @param resource_root: API resource object. 164 @param attrs: optional dictionary of attributes to set. This should only 165 contain r/w attributes. 166 """ 167 self._resource_root = resource_root 168 169 for name, attr in self._get_attributes().iteritems(): 170 object.__setattr__(self, name, None) 171 if attrs: 172 self._set_attrs(attrs, from_json=False)

173

174 - def _set_attrs(self, attrs, allow_ro=False, from_json=True):

175 """ 176 Sets all the attributes in the dictionary. Optionally, allows setting 177 read-only attributes (e.g. when deserializing from JSON) and skipping 178 JSON deserialization of values. 179 """ 180 for k, v in attrs.iteritems(): 181 attr = self._check_attr(k, allow_ro) 182 if attr and from_json: 183 v = attr.from_json(self._get_resource_root(), v) 184 object.__setattr__(self, k, v)

185

186 - def __setattr__(self, name, val):

187 if name not in BaseApiObject._WHITELIST: 188 self._check_attr(name, False) 189 object.__setattr__(self, name, val)

190

191 - def _check_attr(self, name, allow_ro):

192 if name not in self._get_attributes(): 193 raise AttributeError('Invalid property %s for class %s.' % 194 (name, self.__class__.__name__)) 195 attr = self._get_attributes()[name] 196 if not allow_ro and attr and not attr.rw: 197 raise AttributeError('Attribute %s of class %s is read only.' % 198 (name, self.__class__.__name__)) 199 return attr

200

201 - def _get_resource_root(self):

202 return self._resource_root

203

204 - def _update(self, api_obj):

205 """Copy state from api_obj to this object.""" 206 if not isinstance(self, api_obj.__class__): 207 raise ValueError( 208 "Class %s does not derive from %s; cannot update attributes." % 209 (self.__class__, api_obj.__class__)) 210 211 for name in self._get_attributes().keys(): 212 try: 213 val = getattr(api_obj, name) 214 setattr(self, name, val) 215 except AttributeError, ignored: 216 pass

217

218 - def to_json_dict(self, preserve_ro=False):

219 dic = { } 220 for name, attr in self._get_attributes().iteritems(): 221 if not preserve_ro and attr and not attr.rw: 222 continue 223 try: 224 value = getattr(self, name) 225 if attr: 226 dic[name] = attr.to_json(value, preserve_ro) 227 else: 228 dic[name] = value 229 except AttributeError: 230 pass 231 return dic

232

233 - def __str__(self):

234 """ 235 Default implementation of __str__. Uses the type name and the first 236 attribute retrieved from the attribute map to create the string. 237 """ 238 name = self._get_attributes().keys()[0] 239 value = getattr(self, name, None) 240 return "<%s>: %s = %s" % (self.__class__.__name__, name, value)

241 242 @classmethod

243 - def from_json_dict(cls, dic, resource_root):

244 obj = cls(resource_root) 245 obj._set_attrs(dic, allow_ro=True) 246 return obj

247

248 - def _require_min_api_version(self, version):

249 """ 250 Raise an exception if the version of the api is less than the given version. 251 252 @param version: The minimum required version. 253 """ 254 actual_version = self._get_resource_root().version 255 if actual_version < version: 256 raise Exception("API version %s is required but %s is in use." 257 % (version, actual_version))

258

259 -class ApiList(object):

260 """A list of some api object""" 261 LIST_KEY = "items" 262

263 - def __init__(self, objects):

264 self.objects = objects

265

266 - def __str__(self):

267 return "<ApiList>(%d): [%s]" % ( 268 len(self.objects), 269 ", ".join([str(item) for item in self.objects]))

270

271 - def to_json_dict(self):

272 return { ApiList.LIST_KEY : 273 [ x.to_json_dict() for x in self.objects ] }

274

275 - def __len__(self):

276 return self.objects.__len__()

277

278 - def __iter__(self):

279 return self.objects.__iter__()

280

281 - def __getitem__(self, i):

282 return self.objects.__getitem__(i)

283

284 - def __getslice(self, i, j):

285 return self.objects.__getslice__(i, j)

286 287 @staticmethod

288 - def from_json_dict(member_cls, dic, resource_root):

289 json_list = dic[ApiList.LIST_KEY] 290 objects = [ member_cls.from_json_dict(x, resource_root) for x in json_list ] 291 return ApiList(objects)

292

293 294 -class ApiHostRef(BaseApiObject):

295 _ATTRIBUTES = { 296 'hostId' : None, 297 } 298

299 - def __init__(self, resource_root, hostId=None):

300 BaseApiObject.init(self, resource_root, locals())

301

302 - def __str__(self):

303 return "<ApiHostRef>: %s" % (self.hostId)

304

305 -class ApiServiceRef(BaseApiObject):

306 _ATTRIBUTES = { 307 'clusterName' : None, 308 'serviceName' : None, 309 'peerName' : None, 310 } 311

312 - def __init__(self, resource_root, serviceName=None, clusterName=None, 313 peerName=None):

314 BaseApiObject.init(self, resource_root, locals())

315

316 -class ApiClusterRef(BaseApiObject):

317 _ATTRIBUTES = { 318 'clusterName' : None, 319 } 320

321 - def __init__(self, resource_root, clusterName = None):

322 BaseApiObject.init(self, resource_root, locals())

323

324 -class ApiRoleRef(BaseApiObject):

325 _ATTRIBUTES = { 326 'clusterName' : None, 327 'serviceName' : None, 328 'roleName' : None, 329 } 330

331 - def __init__(self, resource_root, serviceName=None, roleName=None, 332 clusterName=None):

333 BaseApiObject.init(self, resource_root, locals())

334

335 -class ApiRoleConfigGroupRef(BaseApiObject):

336 _ATTRIBUTES = { 337 'roleConfigGroupName' : None, 338 } 339

340 - def __init__(self, resource_root, roleConfigGroupName=None):

341 BaseApiObject.init(self, resource_root, locals())

342

343 -class ApiCommand(BaseApiObject):

344 SYNCHRONOUS_COMMAND_ID = -1 345 346 @classmethod

347 - def _get_attributes(cls):

348 if not cls.__dict__.has_key('_ATTRIBUTES'): 349 cls._ATTRIBUTES = { 350 'id' : ROAttr(), 351 'name' : ROAttr(), 352 'startTime' : ROAttr(datetime.datetime), 353 'endTime' : ROAttr(datetime.datetime), 354 'active' : ROAttr(), 355 'success' : ROAttr(), 356 'resultMessage' : ROAttr(), 357 'clusterRef' : ROAttr(ApiClusterRef), 358 'serviceRef' : ROAttr(ApiServiceRef), 359 'roleRef' : ROAttr(ApiRoleRef), 360 'hostRef' : ROAttr(ApiHostRef), 361 'children' : ROAttr(ApiCommand, is_api_list=True), 362 'parent' : ROAttr(ApiCommand), 363 'resultDataUrl' : ROAttr(), 364 } 365 return cls._ATTRIBUTES

366

367 - def __str__(self):

368 return "<ApiCommand>: '%s' (id: %s; active: %s; success: %s)" % ( 369 self.name, self.id, self.active, self.success)

370

371 - def _path(self):

372 return '/commands/%d' % self.id

373

374 - def fetch(self):

375 """ 376 Retrieve updated data about the command from the server. 377 378 @param resource_root: The root Resource object. 379 @return: A new ApiCommand object. 380 """ 381 if self.id == ApiCommand.SYNCHRONOUS_COMMAND_ID: 382 return self 383 384 resp = self._get_resource_root().get(self._path()) 385 return ApiCommand.from_json_dict(resp, self._get_resource_root())

386

387 - def wait(self, timeout=None):

388 """ 389 Wait for command to finish. 390 391 @param timeout: (Optional) Max amount of time (in seconds) to wait. Wait 392 forever by default. 393 @return: The final ApiCommand object, containing the last known state. 394 The command may still be running in case of timeout. 395 """ 396 if self.id == ApiCommand.SYNCHRONOUS_COMMAND_ID: 397 return self 398 399 SLEEP_SEC = 5 400 401 if timeout is None: 402 deadline = None 403 else: 404 deadline = time.time() + timeout 405 406 while True: 407 cmd = self.fetch() 408 if not cmd.active: 409 return cmd 410 411 if deadline is not None: 412 now = time.time() 413 if deadline < now: 414 return cmd 415 else: 416 time.sleep(min(SLEEP_SEC, deadline - now)) 417 else: 418 time.sleep(SLEEP_SEC)

419 420

421 - def abort(self):

422 """ 423 Abort a running command. 424 425 @param resource_root: The root Resource object. 426 @return: A new ApiCommand object with the updated information. 427 """ 428 if self.id == ApiCommand.SYNCHRONOUS_COMMAND_ID: 429 return self 430 431 path = self._path() + '/abort' 432 resp = self._get_resource_root().post(path) 433 return ApiCommand.from_json_dict(resp, self._get_resource_root())

434

435 # 436 # Metrics. 437 # 438 439 -class ApiMetricData(BaseApiObject):

440 """Metric reading data.""" 441 442 _ATTRIBUTES = { 443 'timestamp' : ROAttr(datetime.datetime), 444 'value' : ROAttr(), 445 } 446

447 - def __init__(self, resource_root):

448 BaseApiObject.init(self, resource_root)

449

450 451 -class ApiMetric(BaseApiObject):

452 """Metric information.""" 453 454 _ATTRIBUTES = { 455 'name' : ROAttr(), 456 'context' : ROAttr(), 457 'unit' : ROAttr(), 458 'data' : ROAttr(ApiMetricData), 459 'displayName' : ROAttr(), 460 'description' : ROAttr(), 461 } 462

463 - def __init__(self, resource_root):

464 BaseApiObject.init(self, resource_root)

465

466 # 467 # Activities. 468 # 469 470 -class ApiActivity(BaseApiObject):

471 _ATTRIBUTES = { 472 'name' : ROAttr(), 473 'type' : ROAttr(), 474 'parent' : ROAttr(), 475 'startTime' : ROAttr(), 476 'finishTime' : ROAttr(), 477 'id' : ROAttr(), 478 'status' : ROAttr(), 479 'user' : ROAttr(), 480 'group' : ROAttr(), 481 'inputDir' : ROAttr(), 482 'outputDir' : ROAttr(), 483 'mapper' : ROAttr(), 484 'combiner' : ROAttr(), 485 'reducer' : ROAttr(), 486 'queueName' : ROAttr(), 487 'schedulerPriority' : ROAttr(), 488 } 489

490 - def __init__(self, resource_root):

491 BaseApiObject.init(self, resource_root)

492

493 - def __str__(self):

494 return "<ApiActivity>: %s (%s)" % (self.name, self.status)

495

496 # 497 # Replication 498 # 499 500 -class ApiCmPeer(BaseApiObject):

501 _ATTRIBUTES = { 502 'name' : None, 503 'url' : None, 504 'username' : None, 505 'password' : None, 506 } 507

508 - def __str__(self):

509 return "<ApiPeer>: %s (%s)" % (self.name, self.url)

510

511 -class ApiHdfsReplicationArguments(BaseApiObject):

512 _ATTRIBUTES = { 513 'sourceService' : Attr(ApiServiceRef), 514 'sourcePath' : None, 515 'destinationPath' : None, 516 'mapreduceServiceName' : None, 517 'userName' : None, 518 'numMaps' : None, 519 'dryRun' : None, 520 'schedulerPoolName' : None, 521 'abortOnError' : None, 522 'preservePermissions' : None, 523 'preserveBlockSize' : None, 524 'preserveReplicationCount' : None, 525 'removeMissingFiles' : None, 526 'skipChecksumChecks' : None, 527 }

528

529 -class ApiHdfsReplicationResult(BaseApiObject):

530 _ATTRIBUTES = { 531 'progress' : ROAttr(), 532 'counters' : ROAttr(), 533 'numBytesDryRun' : ROAttr(), 534 'numFilesDryRun' : ROAttr(), 535 'numFilesExpected' : ROAttr(), 536 'numBytesExpected' : ROAttr(), 537 'numFilesCopied' : ROAttr(), 538 'numBytesCopied' : ROAttr(), 539 'numFilesSkipped' : ROAttr(), 540 'numBytesSkipped' : ROAttr(), 541 'numFilesDeleted' : ROAttr(), 542 'numFilesCopyFailed' : ROAttr(), 543 'numBytesCopyFailed' : ROAttr(), 544 'setupError' : ROAttr(), 545 'jobId' : ROAttr(), 546 'jobDetailsUri' : ROAttr(), 547 'dryRun' : ROAttr(), 548 }

549

550 -class ApiHiveTable(BaseApiObject):

551 _ATTRIBUTES = { 552 'database' : None, 553 'tableName' : None, 554 } 555

556 - def __str__(self):

557 return "<ApiHiveTable>: %s, %s" % (self.database, self.tableName)

558

559 -class ApiHiveReplicationArguments(BaseApiObject):

560 _ATTRIBUTES = { 561 'sourceService' : Attr(ApiServiceRef), 562 'tableFilters' : Attr(ApiHiveTable), 563 'exportDir' : None, 564 'force' : None, 565 'replicateData' : None, 566 'hdfsArguments' : Attr(ApiHdfsReplicationArguments), 567 'dryRun' : None, 568 }

569

570 -class ApiHiveReplicationResult(BaseApiObject):

571 _ATTRIBUTES = { 572 'tables' : ROAttr(ApiHiveTable), 573 'errors' : ROAttr(), 574 'dataReplicationResult' : ROAttr(ApiHdfsReplicationResult), 575 'dryRun' : ROAttr(), 576 }

577

578 -class ApiReplicationCommand(ApiCommand):

579 @classmethod

580 - def _get_attributes(cls):

581 if not cls.__dict__.has_key('_ATTRIBUTES'): 582 attrs = { 583 'hdfsResult' : ROAttr(ApiHdfsReplicationResult), 584 'hiveResult' : ROAttr(ApiHiveReplicationResult), 585 } 586 attrs.update(ApiCommand._get_attributes()) 587 cls._ATTRIBUTES = attrs 588 return cls._ATTRIBUTES

589

590 -class ApiReplicationSchedule(BaseApiObject):

591 _ATTRIBUTES = { 592 'startTime' : Attr(datetime.datetime), 593 'endTime' : Attr(datetime.datetime), 594 'interval' : None, 595 'intervalUnit' : None, 596 'paused' : None, 597 'hdfsArguments' : Attr(ApiHdfsReplicationArguments), 598 'hiveArguments' : Attr(ApiHiveReplicationArguments), 599 'alertOnStart' : None, 600 'alertOnSuccess' : None, 601 'alertOnFail' : None, 602 'alertOnAbort' : None, 603 'id' : ROAttr(), 604 'nextRun' : ROAttr(datetime.datetime), 605 'history' : ROAttr(ApiReplicationCommand), 606 }

607

608 # 609 # Configuration helpers. 610 # 611 612 -class ApiConfig(BaseApiObject):

613 _ATTRIBUTES = { 614 'name' : None, 615 'value' : None, 616 'required' : ROAttr(), 617 'default' : ROAttr(), 618 'displayName' : ROAttr(), 619 'description' : ROAttr(), 620 'relatedName' : ROAttr(), 621 'validationState' : ROAttr(), 622 'validationMessage' : ROAttr(), 623 } 624

625 - def __init__(self, resource_root, name=None, value=None):

626 BaseApiObject.init(self, resource_root, locals())

627

628 - def __str__(self):

629 return "<ApiConfig>: %s = %s" % (self.name, self.value)

630

631 -class ApiImpalaQuery(BaseApiObject):

632 _ATTRIBUTES = { 633 'queryId' : ROAttr(), 634 'queryState' : ROAttr(), 635 'queryType' : ROAttr(), 636 'statement' : ROAttr(), 637 'database' : ROAttr(), 638 'rowsProduced' : ROAttr(), 639 'coordinator' : ROAttr(ApiHostRef), 640 'user' : ROAttr(), 641 'startTime' : ROAttr(datetime.datetime), 642 'endTime' : ROAttr(datetime.datetime), 643 'detailsAvailable' : ROAttr(), 644 'attributes' : ROAttr(), 645 'durationMillis' : ROAttr() 646 } 647

648 - def __str__(self):

649 return "<ApiImpalaQuery>: %s" % (self.queryId)

650

651 652 -class ApiImpalaQueryResponse(BaseApiObject):

653 654 _ATTRIBUTES = { 655 'queries' : ROAttr(ApiImpalaQuery), 656 'warnings' : ROAttr() 657 }

658

659 -class ApiImpalaQueryDetailsResponse(BaseApiObject):

660 _ATTRIBUTES = { 661 'details' : ROAttr() 662 } 663

664 - def __str__(self):

665 return "<AipImpalaQueryDetailsResponse> %s" % self.details

666

667 -class ApiImpalaCancelResponse(BaseApiObject):

668 _ATTRIBUTES = { 669 'warning' : ROAttr() 670 } 671

672 - def __str__(self):

673 return "<ApiImpalaCancelResponse> %s" % self.warning

674

675 -def config_to_api_list(dic):

676 """ 677 Converts a python dictionary into a list containing the proper 678 ApiConfig encoding for configuration data. 679 680 @param dic Key-value pairs to convert. 681 @return JSON dictionary of an ApiConfig list (*not* an ApiList). 682 """ 683 config = [ ] 684 for k, v in dic.iteritems(): 685 config.append({ 'name' : k, 'value': v }) 686 return { ApiList.LIST_KEY : config }

687

688 -def config_to_json(dic):

689 """ 690 Converts a python dictionary into a JSON payload. 691 692 The payload matches the expected "apiConfig list" type used to update 693 configuration parameters using the API. 694 695 @param dic Key-value pairs to convert. 696 @return String with the JSON-encoded data. 697 """ 698 return json.dumps(config_to_api_list(dic))

699

700 -def json_to_config(dic, full = False):

701 """ 702 Converts a JSON-decoded config dictionary to a python dictionary. 703 704 When materializing the full view, the values in the dictionary will be 705 instances of ApiConfig, instead of strings. 706 707 @param dic JSON-decoded config dictionary. 708 @param full Whether to materialize the full view of the config data. 709 @return Python dictionary with config data. 710 """ 711 config = { } 712 for entry in dic['items']: 713 k = entry['name'] 714 if full: 715 config[k] = ApiConfig.from_json_dict(entry, None) 716 else: 717 config[k] = entry.get('value') 718 return config

719

Source Code for Module cm_api.endpoints.types