Skip to main content

CloudFoundry 的客户端库

项目描述

https://img.shields.io/pypi/v/cloudfoundry-client.svg https://img.shields.io/github/license/antechrestos/cf-python-client.svg

cf-python-client 存储库包含 Cloud Foundry 的 Python 客户端库。

安装

支持的版本

  • 从版本1.11.0开始,将不再支持比 python 3.6.0更早的版本。此最新版本于 2016 年底发布。对于仍在使用 python 2.7 的用户,到 2020 年底将不再支持它,所有库将停止支持它。

  • 从版本1.25.0开始,将不再支持比 python 3.7.0更早的版本。

官方文档

从点子

$ pip install cloudfoundry-client

从来源

要构建库运行:

$ python setup.py install

运行客户端

要运行客户端,请输入以下命令:

$ cloudfoundry-client

这将向您解释客户端的工作方式。在第一次执行时,它会询问您有关您想要访问的平台的信息(网址、登录名等)。请注意,您的凭据不会保存在您的磁盘上:只会保留令牌以供进一步使用。

在您的代码中使用客户端

您可以构建客户端并在您的代码中使用它

客户

实例化客户端,没有比这更容易的了

from cloudfoundry_client.client import CloudFoundryClient
target_endpoint = 'https://somewhere.org'
proxy = dict(http=os.environ.get('HTTP_PROXY', ''), https=os.environ.get('HTTPS_PROXY', ''))
client = CloudFoundryClient(target_endpoint, proxy=proxy, verify=False)
# init with user credentials
client.init_with_user_credentials('login', 'password')
# init with refresh token (that will retrieve a fresh access token)
client.init_with_token('refresh-token')
# init with access and refresh token (if the above method is not convenient)
client.refresh_token = 'refresh-token'
client._access_token = 'access-token'

如果您拥有带有重定向功能的专用 oauth 应用程序,也可以使用 oauth 代码流对其进行实例化

from flask import request
from cloudfoundry_client.client import CloudFoundryClient
target_endpoint = 'https://somewhere.org'
proxy = dict(http=os.environ.get('HTTP_PROXY', ''), https=os.environ.get('HTTPS_PROXY', ''))
client = CloudFoundryClient(target_endpoint, proxy=proxy, verify=False, client_id='my-client-id', client_secret='my-client-secret')

@app.route('/login')
def login():
    global client
    return redirect(client.generate_authorize_url('http://localhost:9999/code', '666'))

@app.route('/code')
def code():
    global client
    client.init_authorize_code_process('http://localhost:9999/code', request.args.get('code'))

然后您可以按如下方式使用它:

for organization in client.v2.organizations:
    print(organization['metadata']['guid'])

API V2

实体

api V2 调用返回的实体(组织空间应用程序..)是可导航的,即您可以调用与xxx_url实体属性关联的方法(请注意,如果属性的名称以列表结尾,它将被解释为对象列表. 否则你会得到一个单一的实体)。

for organization in client.v2.organizations:
    for space in organization.spaces(): # perform a GET on spaces_url attribute
        organization_reloaded = space.organization()  # perform a GET on organization_url attribute
Application 对象提供了更多的方法,例如
  • 实例

  • 统计数据

  • 开始

  • 停止

  • 概括

例如,您可以获得所有摘要,如下所示:

要不然:

for app in client.v2.apps:
    print(app.summary())

可用的经理

到目前为止,可用的已实施管理器有:

  • 服务计划

  • service_plan_visibilities

  • 服务实例

  • 服务键

  • 服务绑定

  • service_brokers

  • 应用

  • 事件

  • 构建包

  • 组织

  • 空间

  • 服务

  • 路线

  • 共享域

  • 私人域名

  • 安全组

请注意,即使在导航时到达不存在的实体管理器,也会执行获取,并且您将获得预期的实体。例如,事件实体管理器尚未实现,但您可以执行

for app in client.v2.apps:
    for event in app.events():
        handle_event_object()

所有管理器都提供以下方法:

  • list(**kwargs):根据给定的过滤参数返回实体上的迭代器

  • get_first(**kwargs):根据给定的参数返回第一个匹配的实体。如果没有返回,则返回`None

  • get :对实体执行GET 。如果无法找到实体,则会由于 http NOT FOUND响应状态而引发异常

  • __iter__:对管理器本身的迭代。无过滤器列表的别名

  • __getitem__ : get操作的别名

  • _create:创建操作。由于是泛型操作(只接受一个dict对象),所以这个操作是受保护的

  • _update:更新操作。由于它是一个通用操作(只需要一个资源 id 和一个dict对象),这个操作是受保护的

  • _remove:删除操作。此操作保持受保护。

# Assume you have an organization named `test-org` with a guid of `test-org-guid`
org_get = client.v2.organizations.get('test-org-guid')
org_get_first = client.v2.organizations.get_first(**{'name': 'test-org'})
org_from_list = list(client.v2.organizations.list(**{'name': 'test-org'}))[0]
assert org_get == org_get_first == org_from_list

# You can also specify multiple values for a query parameter.
for organization in client.v2.organizations.list(**{'name': ['org1', 'org2']}):
    print(organization['metadata']['guid'])

# Order and Paging parameters are also supported.
query = {
    'order-by': 'name',
    'order-direction': 'desc',
    'results-per-page': 100
}
for organization in client.v2.organizations.list(**query):
    print(organization['entity']['name'])

API V3

实体

API V3 返回的实体通过提供对具有链接本身名称的对象的调用来调用脚本链接。让我们用下一个代码来解释它

for app in client.v3.apps.list(space_guids='space_guid'):
  for task in app.tasks():
      print('Task %s' % task['guid'])
  app.stop()
  space = app.space()

另一个例子:

app = client.v3.apps['app-guid']
for task in app.tasks():
    task.cancel()
for task in client.v3.tasks.list(app_guids=['app-guid-1', 'app-guid-2']):
    task.cancel()

当 API 支持时,父实体可以包含在单个调用中。包含的实体替换了上面提到的链接。以下代码片段向 API 发出三个请求,以获取应用、空间和组织数据:

app = client.v3.apps.get("app-guid")
print("App name: %s" % app["name"])
space = app.space()
print("Space name: %s" % space["name"])
org = space.organization()
print("Org name: %s" % org["name"])

通过仅更改第一行,单个请求获取所有数据。从应用程序到空间和空间到组织的导航保持不变。

app = client.v3.apps.get("app-guid", include="space.organization")

API V3 上可用的管理器有:

  • 应用

  • 构建包

  • 特征标志

  • 隔离段

  • 工作

  • 组织

  • 组织配额

  • 流程

  • 角色

  • 安全组

  • service_brokers

  • service_credential_bindings

  • 服务实例

  • service_offerings

  • 服务计划

  • 空间

  • 任务

管理器提供与 V2 管理器相同的方法,但有以下区别:

  • get(**kwargs):支持传递给 API 的关键字参数,例如“include”

联网

策略服务器

目前我们只实施了网络策略

for policy in client.network.v1.external.policies.list():
  print('destination protocol = {}'.format(policy['destination']['protocol']))
  print('destination from port = {}'.format(policy['destination']['ports']['start']))
  print('destination to port = {}'.format(policy['destination']['ports']['end']))

API V3 上可用的管理器有:

  • 政策

该经理提供:

  • list(**kwargs):根据给定的过滤参数返回实体上的迭代器

  • __iter__:对管理器本身的迭代。无过滤器列表的别名

  • _create:创建操作。由于是泛型操作(只接受一个dict对象),所以这个操作是受保护的

  • _remove:删除操作。此操作保持受保护。

应用程序日志

应用程序的最近日志可以得到如下:

app = client.v2.apps['app-guid']
for log in app.recent_logs():
    print(log)

也可以使用 websocket 流式传输日志,如下所示:

app = client.v2.apps['app-guid']
for log in app.stream_logs():
    # read message infinitely (use break to exit... it will close the underlying websocket)
    print(log)
# or
for log in client.doppler.stream_logs('app-guid'):
    # read message infinitely (use break to exit... it will close the underlying websocket)
    print(log)

日志也可以直接从 RLP 网关流式传输:

import asyncio
from cloudfoundry_client.client import CloudFoundryClient

target_endpoint = 'https://somewhere.org'
proxy = dict(http=os.environ.get('HTTP_PROXY', ''), https=os.environ.get('HTTPS_PROXY', ''))
rlp_client = CloudFoundryClient(target_endpoint, client_id='client_id', client_secret='client_secret', verify=False)
# init with client credentials
rlp_client.init_with_client_credentials()

async def get_logs_for_app(rlp_client, app_guid):
    async for log in rlp_client.rlpgateway.stream_logs(app_guid,
                                                       params={'counter': '', 'gauge': ''},
                                                       headers={'User-Agent': 'cf-python-client'})):
        print(log)

loop = asyncio.get_event_loop()
loop.create_task(get_logs_for_app(rlp_client, "app_guid"))
loop.run_forever()
loop.close()

命令行界面

客户端带有命令行界面。运行cloudfoundry-client命令。在第一次执行时,它会询问您有关目标平台和您的凭证的信息(不要担心它们没有保存)。之后,您可以通过运行cloudfoundry-client -h获得帮助

操作(实验)

目前唯一实现的操作是推送操作。

from cloudfoundry_client.operations.push.push import PushOperation
operation = PushOperation(client)
operation.push(client.v2.spaces.get_first(name='My Space')['metadata']['guid'], path)

问题和贡献

请提交问题/拉取请求。

您可以通过这样做来运行测试。在项目目录中:

$ export PYTHONPATH=main
$ python -m unittest discover test
# or even
$ python setup.py test