Skip to main content

A limited OpenStack Identity API v3 client in Python (with fewer dependencies)

Project description

keystone-light implements a Python interface to a very limited subset of the OpenStack Identity API v3.

Initial goal: access to OpenStack Swift, using the Identity API v3, but with a lot fewer dependencies.

As of this writing, the python-keystoneclient requires keystoneauth1 and oslo.*, which in turn require some more. We only require the ubiquitous requests (and PyYAML), which you generally already have installed anyway.

Example usage

#!/usr/bin/env python3
from urllib.parse import urljoin

import requests
from keystone_light import Cloud, CloudsYamlConfig, PermissionDenied

def get_projects(cloud):
    "Yields projects, sorted by domain and project name"
    domains = cloud.get_domains()
    for domain in sorted(domains, key=(lambda x:
        if == 'Default':
            # print('WARN: skipping domain Default (fixme?)')

        projects = domain.get_projects()
        for project in sorted(projects, key=(lambda x:
            project.domain = domain
            yield project

def _give_us_project_perms_through_admin_group(project):
    Make sure we are in the *-admin group. Make sure the *-admin
    group has permissions on the project.
    cloud =
    dom_admin_group = project.domain.get_admin_group()

    # First check if we're member of the group at all.
    token = cloud.get_system_token()
    auth_headers = {'X-Auth-Token': str(token)}
        # FIXME: Undocumented access to system_token!
        user_id =['user']['id']
        assert user_id and isinstance(user_id, str), user_id
    except KeyError:
        raise ValueError('missing',

    # Are we in the *-admin group?
    url = urljoin(
  , user_id=user_id))
    out = requests.head(url, headers=auth_headers)
    if out.status_code == 404:
        # Add us to the group.
        out = requests.put(url, headers=auth_headers)
        assert out.status_code == 204, (
            'PUT', url, out.status_code, out.text)
        # Double check.
        out = requests.head(url, headers=auth_headers)
    assert out.status_code == 204, (
        'HEAD', url, out.status_code, out.text)

    # Grant *-admin power to the project.
    admin_role = cloud.get_role(name='admin')  # or 'reader'
    url = urljoin(
    out = requests.put(url, headers=auth_headers)
    assert out.status_code in (201, 204), (
        'PUT', url, out.status_code, out.text)

def get_swift_stat_ensuring_permissions(project):
    "Get Swift v1 stat on a project (previously: tenant)"
        stat = project.get_swift().get_stat()
    except PermissionDenied:
        # We don't have permission to access the project? Upgrade the
        # permissions and try again.
        return stat

    # Try again. Should succeed now, with the added permissions.
        stat = project.get_swift().get_stat()
    except PermissionDenied as e:
        raise MyPermissionDenied(
            'EPERM on {domain}.{project}: {exc} {exc_args}'.format(
                exc=e.__class__.__name__, exc_args=e.args)) from e
        return stat

# Take config from ~/.config/openstack/clouds.yaml and select
# 'my-cloud-admin', like the openstack(1) --os-cloud option.
config = CloudsYamlConfig('my-cloud-admin')
cloud = Cloud(config)
for project in get_projects(cloud):
    swift_stat = get_swift_stat_ensuring_permissions(project)
    print('{:15s} {:23s} {:21d} B ({} objects, {} containers)'.format([0:15],,

Example output

$ python3
domainx         project                  3489 B (2 objects, 1 containers)
domainx         otherproject       1455042022 B (267 objects, 1 containers)

Swift Example usage

from keystone_light import Cloud, DirectConfig

SWIFT_CONTAINER = 'some-container'

config = DirectConfig(KEYSTONE_URI)
project = Cloud(config).get_current_project()
assert project.get_fullname() == SWIFT_PROJECT, project.get_fullname()

swift = project.get_swift()
container = swift.get_container(SWIFT_CONTAINER)

# (Re-)upload file:
filename = ('bloblet.bin' if False else 'blobzilla.bin')
with open(filename, 'rb') as fp:
    except FileNotFoundError:
    # TIP: Use ChunkIteratorIOBaseWrapper(fp) if the input file
    # is a pipe/stream.
    container.put(filename, fp)

# Download file:
filename2 = '{}.retrieved'.format(filename)
with container.get(filename) as response, \
        open(filename2, 'wb') as fp:
    for chunk in response.iter_content(chunk_size=8192):

# Check and compare:
with open(filename, 'rb') as fp, \
        open(filename2, 'rb') as fp2:
    buf = buf2 = True
    while buf and buf2:
        buf =
        buf2 =
        assert buf == buf2
    assert buf == buf2

And an example with timing:

from timeit import timeit

# ...

# Download file:
filename2 = '{}.retrieved'.format(filename)
def _get():
    with container.get(filename) as response, \
            open(filename2, 'wb') as fp:
        for chunk in response.iter_content(chunk_size=8192):
print('{:7.3f} GET'.format(timeit(number=1, stmt=_get)))

Project details

Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

keystone-light-1.3.tar.gz (15.5 kB view hashes)

Uploaded Source

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page