aliyunsdkros源码及相关API
参考文章：
源码下载：
一、目录结构
--aliyunsdkros
--request
--v20150901 # 版本
--init.py
--AbandonStackRequest.py 废弃资源栈
--CreateStacksRequest.py 创建资源栈
--DeleteStackRequest.py 删除资源栈
--DescribeEventsRequest.py 查询事件列表
--DescribeRegionsRequest.py 查询Region列表
--DescribeResourceDetailRequest.py 查询资源详情
--DescribeResourcesRequest.py 查询资源列表
--DescribeResourceTypeDetailRequest.py 查询资源类型详情
--DescribeResourceTypesRequest.py 查询资源类型列表
--DescribeResourceTypeTemplateRequest.py 查询资源模板示例
--DescribeStackDetailRequest.py 查询资源栈详情
--DescribeStacksRequest.py 查询资源栈列表
--DescribeTemplateRequest.py 查询模板
--DoActionsRequest.py
--InquiryStackRequest.py
--PreviewStackRequest.py 预览资源栈
--UpdateStackRequest.py 更新资源栈
--ValidateTemplateRequest.py 验证模板合法性
--init.py
--init.py二、API概览
资源栈相关接口
| URI | Method | 描述 |
|---|---|---|
| /stacks | POST | 创建资源栈。 |
| /stacks | GET | 查询资源栈列表。 |
| /stacks/{StackName}/{StackId} | GET | 查询资源栈信息。 |
| /stacks/{StackName}/{StackId} | PUT | 更新资源栈。 |
| /stacks/{StackName}/{StackId} | DELETE | 删除资源栈。 |
| /stacks/{StackName}/{StackId}/abandon | DELETE | 废弃资源栈。 |
| /stacks/preview | POST | 预览资源栈。 |
资源相关接口
| URI | Method | 描述 |
|---|---|---|
| /stacks/{StackName}/{StackId}/resources | GET | 查询资源列表。 |
| /stacks/{StackName}/{StackId}/resources/{ResourceName} | GET | 查询资源信息。 |
| /resource_types | GET | 查询资源类型列表。 |
| /resource_types/{TypeName} | GET | 查询资源类型信息。 |
| /resource_types/{TypeName}/template | GET | 查询资源类型模板信息。 |
模板相关接口
| URI | Method | 描述 |
|---|---|---|
| /stacks/{StackName}/{StackId}/template | GET | 查询模板信息。 |
| /validate | POST | 验证模板信息。 |
其他接口
| URI | Method | 描述 |
|---|---|---|
| /stacks/{StackName}/{StackId}/events | GET | 查询事件列表。 |
| /regions | GET | 查询 Region 列表。 |
三、源码
aliyunsdkros模块主要封装ROS资源编排服务提供的API接口。
该模块下的Request类均继承自aliyunsdkcore.request.RoaRequest，RoaRequest则继承自同AcsRequest，所以先贴出AcsReques和RoaRequest的源码。
1、AcsRequest
class AcsRequest:
"""
Acs request base class. This class wraps up common parameters for a request.
Acs请求类的基类，这个类为请求类封装通用参数。
"""
# 定义AcsRequest为抽象基类
__metaclass__ = abc.ABCMeta
def __init__(self, product, version=None,
action_name=None,
location_service_code=None,
location_endpoint_type='openAPI',
accept_format=None,
protocol_type=None,
method=None):
"""
:param product: 产品
:param version: 版本
:param action_name: 指令名/操作名
:param params: 参数
:param resource_owner_account:
:param protocol_type: 协议类型
:param accept_format: ？
:param method: 请求方法
:return:
"""
# 初始化参数
self._version = version
self._product = product
self._action_name = action_name
self._protocol_type = protocol_type
# 未指定协议类型则使用默认：protocol_type.HTTP（http协议）
if self._protocol_type is None:
self._protocol_type = _default_protocol_type
self._accept_format = accept_format
self._params = {}
self._method = method
self._header = {}
self._body_params = {}
self._uri_pattern = None
self._uri_params = None
self._content = None
self._location_service_code = location_service_code
self._location_endpoint_type = location_endpoint_type
self.add_header('x-sdk-invoke-type', 'normal')
def add_query_param(self, k, v):
self._params[k] = v
def add_body_params(self, k, v):
self._body_params[k] = v
def get_body_params(self):
return self._body_params
def get_uri_pattern(self):
return self._uri_pattern
def get_uri_params(self):
return self._uri_params
def get_product(self):
return self._product
def get_version(self):
return self._version
def get_action_name(self):
return self._action_name
def get_accept_format(self):
return self._accept_format
def get_protocol_type(self):
return self._protocol_type
def get_query_params(self):
return self._params
def get_method(self):
return self._method
def set_uri_pattern(self, pattern):
self._uri_pattern = pattern
def set_uri_params(self, params):
self._uri_params = params
def set_method(self, method):
self._method = method
def set_product(self, product):
self._product = product
def set_version(self, version):
self._version = version
def set_action_name(self, action_name):
self._action_name = action_name
def set_accept_format(self, accept_format):
self._accept_format = accept_format
def set_protocol_type(self, protocol_type):
self._protocol_type = protocol_type
def set_query_params(self, params):
self._params = params
def set_content(self, content):
# content：字节数组
self._content = content
def get_content(self):
# 返回字节数组
return self._content
def get_headers(self):
# 返回dict
return self._header
def set_headers(self, headers):
# headers：dict
self._header = headers
def add_header(self, k, v):
self._header[k] = v
def set_user_agent(self, agent):
self.add_header('User-Agent', agent)
def set_location_service_code(self, location_service_code):
self._location_service_code = location_service_code
def get_location_service_code(self):
return self._location_service_code
def get_location_endpoint_type(self):
return self._location_endpoint_type
def set_content_type(self, content_type):
self.add_header("Content-Type", content_type)
# 定义抽象方法
@abc.abstractmethod
def get_style(self):
pass
@abc.abstractmethod
def get_url(self, region_id, ak, secret):
pass
@abc.abstractmethod
def get_signed_header(self, region_id, ak, secret):
pass
2、RoaRequest
class RoaRequest(AcsRequest):
"""
Class to compose an ROA style request with.
具有Roa（面向资源）风格的Request类
"""
def __init__(
self,
product,
version,
action_name,
location_service_code=None,
location_endpoint_type='openAPI',
method=None,
headers=None,
uri_pattern=None,
path_params=None,
protocol=None):
"""
:param product: String 字符串, 必需
:param version: String 字符串, 必需
:param action_name: String 字符串, 必需
:param method: String 字符串
:param headers: Dict 字典
:param uri_pattern: String 字符串
:param path_params: Dict 字典
:param protocol: String 字符串
:return:
"""
AcsRequest.__init__(
self,
product,
version,
action_name,
location_service_code,
location_endpoint_type,
ft.RAW,
protocol,
method)
# 设置默认风格：ROA
self._style = STYLE_ROA
self._method = method
if headers:
self._header = headers
self._uri_pattern = uri_pattern
self._path_params = path_params
def get_style(self):
return self._style
def get_path_params(self):
return self._path_params
def set_path_params(self, path_params):
self._path_params = path_params
def add_path_param(self, k, v):
if self._path_params is None:
self._path_params = {}
self._path_params[k] = v
def _get_sign_params(self):
req_params = self.get_query_params()
if req_params is None:
req_params = {}
self.add_header("x-acs-version", self.get_version())
return req_params
# 计算签名
def get_signed_header(self, region_id, ak, secret):
"""
Generate signed header
:param region_id: String
:param ak: String
:param secret: String
:return: Dict
"""
sign_params = self._get_sign_params()
if self.get_content() is not None:
md5_str = md5_tool.get_md5_base64_str(self.get_content())
self.add_header('Content-MD5', md5_str)
signed_headers = roa_signer.get_signature_headers(
sign_params,
ak,
secret,
self.get_accept_format(),
self.get_headers(),
self.get_uri_pattern(),
self.get_path_params(),
self.get_method())
return signed_headers
# 获取url
def get_url(self, region_id, ak=None, secret=None):
"""
Compose request url without domain
:param region_id: String
:return: String
"""
sign_params = self.get_query_params()
# if region_id not in sign_params.keys():
# sign_params['RegionId'] = region_id
url = roa_signer.get_url(
self.get_uri_pattern(),
sign_params,
self.get_path_params())
return url3、ros的Request.py
ros的Request类url模板、操作名、各自的方法不同外，其余均类似，不做赘述。此处以AbandonStackRequest为例：
class AbandonStackRequest(RoaRequest):
def __init__(self):
# 初始化：产品、版本、操作名
RoaRequest.__init__(self, 'ROS', '2015-09-01', 'AbandonStack')
# 设置url模板
self.set_uri_pattern('/stacks/[StackName]/[StackId]/abandon')
# 设置请求方法
self.set_method('DELETE')
def get_StackName(self):
return self.get_path_params().get('StackName')
def set_StackName(self,StackName):
self.add_path_param('StackName',StackName)
def get_StackId(self):
return self.get_path_params().get('StackId')
def set_StackId(self,StackId):
self.add_path_param('StackId',StackId)
