Deployment(id: int = 0, project_id: int = 0, user_id: int = 0, name: str = '', service_name: str = '', model_inst_id: int = 0, serverless_infer_id: int = 0, deploy_model: str = '', route_path: str = '', token: str = '', token_url: str = '', status: int = 0, infer_lock: int = 0, min_pod: int = 0, max_pod: int = 0, create_time: str = '', update_time: str = '', deploy_time: str = '', is_del: int = 0)
call_service(self, data=None)
调用服务
Source code in deepwisdom/models/deployment.py
def call_service(self, data: Any = None) -> Any:
"""调用服务
Args:
data (Any): 服务调用入参
Returns:
Any: 响应body
"""
if data is None:
api_desc = self.get_service_api(self.id)
data = api_desc.post_example
header = {
"Authorization": "Bearer {}".format(self.token),
"Content-Type": "application/json; charset=utf-8",
}
rsp = self._client.raw_request(
"post", self.token_url, json=data, headers=header)
return rsp
create(project=None, name='Untitled Service', gpu_mem=2, memory_limit=3, max_pod=1, min_pod=1)
classmethod
创建服务
Source code in deepwisdom/models/deployment.py
@classmethod
def create(cls, project: Optional[Project] = None, name: str = "Untitled Service", gpu_mem: int = 2,
memory_limit: int = 3, max_pod: int = 1, min_pod: int = 1) -> "Deployment":
"""
创建服务
Args:
project (Project): 项目对象
name (str): 名称
gpu_mem (int): gpu显存限制GB
memory_limit (int): GB 内存限制
max_pod (int): 最大pod数
min_pod (int): 最小pod数
Returns:
"""
req = CreateDeploymentRequest(project_id=project.project_id, model_inst_id=project.recommended_model().model_id,
name=name, gpu_mem=gpu_mem, memory_limit=memory_limit, max_pod=max_pod,
min_pod=min_pod)
return cls.create_from_req(req)
create_from_req(options)
classmethod
创建服务
Source code in deepwisdom/models/deployment.py
@classmethod
def create_from_req(cls, options: CreateDeploymentRequest) -> "Deployment":
"""创建服务
Args:
options (CreateDeploymentRequest): 服务创建请求参数
Returns:
[Deployment]: 返回具体的服务对象
"""
data = {}
data.update(options)
rsp = cls._client._post(API_URL.DEPLOY_CREATE_SERVICE, data)
if "data" in rsp and "ret" in rsp["data"] and len(rsp["data"]["ret"]) == 1:
deployment = cls.from_server_data(rsp['data']["ret"][0])
# 等待服务部署完成
while True:
status = deployment.get_service_status()
if status == SVR_STATUS.FAIL:
raise err.CreateDeploymentError
if status == SVR_STATUS.RUNNING:
return deployment
else:
raise err.CreateDeploymentError
delete_deployments(service_ids)
classmethod
删除服务
Source code in deepwisdom/models/deployment.py
@classmethod
def delete_deployments(cls, service_ids: List[int]) -> bool:
"""删除服务
Args:
service_ids (int64): 服务id
"""
data = {
"svc_ids": service_ids
}
rsp = cls._client._delete(API_URL.DEPLOY_DELETE_DEPLOYMENT, data)
if rsp['code'] == 200 and rsp['message'] == 'ok':
return True
return False
get_deployment_log(svc_id)
classmethod
获取服务日志
Source code in deepwisdom/models/deployment.py
@classmethod
def get_deployment_log(cls, svc_id: int) -> str:
"""获取服务日志
Args:
svc_id (int): 服务id
Returns:
str: 日志内容
"""
data = {
"service_id": svc_id
}
rsp = cls._client._get(API_URL.DEPLOY_GET_DEPLOYMENT_LOG, data)
if "data" in rsp:
return rsp['data']
return None
get_service(service_id, **kwargs)
classmethod
获取服务详情
Source code in deepwisdom/models/deployment.py
@classmethod
def get_service(cls, service_id, **kwargs) -> "Deployment":
"""获取服务详情
Args:
service_id (int): 服务id
Returns:
Deployment: 服务详情
"""
data = {
"service_id": service_id
}
rsp = cls._server_data(API_URL.DEPLOY_GET_SERVICE_DETAIL, data)
return cls.from_server_data(rsp)
get_service_api(svc_id)
classmethod
获取服务api
Source code in deepwisdom/models/deployment.py
@classmethod
def get_service_api(cls, svc_id: int) -> ServiceApiInfo:
"""获取服务api
Args:
svc_id (int): 服务id
Returns:
ServiceApiInfo: api详情
"""
data = {
"svc_id": svc_id
}
rsp = cls._client._get(API_URL.DEPLOY_GET_SERVICE_API, data)
if "data" in rsp:
checked = ServiceApiInfo._converter.check(rsp['data'])
filtered = Deployment._filter_data(checked)
return ServiceApiInfo(**filtered)
return None
get_service_status(self)
获取服务状态
Source code in deepwisdom/models/deployment.py
def get_service_status(self) -> Int:
"""获取服务状态
Returns:
Int: 服务状态:1-创建,2-调度中,3-调度成功,4-执行中,5-创建失败,6-挂起,7-取消,8-删除中,9- 已删除
"""
svc = self.get_service(self.id)
return svc.status
list_deployments(project_id, **kwargs)
classmethod
获取服务部署列表
Source code in deepwisdom/models/deployment.py
@classmethod
def list_deployments(cls, project_id: int, **kwargs) -> List["DeploymentListMember"]:
"""获取服务部署列表
Args:
project_id (int): 项目id
Returns:
List[DeploymentListMember]: 服务列表
"""
data = {
"project_id": project_id
}
server_data = cls._server_data(API_URL.DEPLOY_LIST_DEPLOYMENTS, data)
# return [cls.get_deployment_detail(item["service_id"]) for item in server_data]
return [DeploymentListMember(**item) for item in server_data]
rename_deployment(svc_id, svc_name)
classmethod
重命名服务
| Parameters: |
svc_id (int) – 服务id
svc_name (str) – 新服务名称
|
Source code in deepwisdom/models/deployment.py
@classmethod
def rename_deployment(cls, svc_id: int, svc_name: str):
"""重命名服务
Args:
svc_id (int): 服务id
svc_name (str): 新服务名称
Returns:
Deployment: 服务详情
"""
data = {
"svc_id": svc_id,
"svc_name": svc_name
}
rsp = cls._client._patch(API_URL.DEPLOY_RENAME_DEPLOYMENT, data)
if rsp['code'] == 200 and rsp['message'] == 'ok':
return cls.get_deployment_detail(svc_id)
return None
resident_deployment(svc_id, min_pod)
classmethod
切换服务常驻状态
Source code in deepwisdom/models/deployment.py
@classmethod
def resident_deployment(cls, svc_id: int, min_pod: int):
"""切换服务常驻状态
Args:
svc_id (int64): 服务id
min_pod (int): 最少pod数,大于0为服务常驻,等于0为非常驻
Returns:
Deployment: 服务详情
"""
data = {
"svc_id": svc_id,
"min_pod": min_pod
}
rsp = cls._client._post(API_URL.DEPLOY_RESIDENT_DEPLOYMENT, data)
if rsp['code'] == 200 and rsp['message'] == 'ok':
return cls.get_deployment_detail(svc_id)
return None