Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
"""
response = client.api.images.post(json=config)
if wait:
return client.operations.wait_for_operation(
response.json()['operation'])
return response.json()['operation']
class Image(model.Model):
"""A LXD Image."""
aliases = model.Attribute(readonly=True)
auto_update = model.Attribute(optional=True)
architecture = model.Attribute(readonly=True)
cached = model.Attribute(readonly=True)
created_at = model.Attribute(readonly=True)
expires_at = model.Attribute(readonly=True)
filename = model.Attribute(readonly=True)
fingerprint = model.Attribute(readonly=True)
last_used_at = model.Attribute(readonly=True)
properties = model.Attribute()
public = model.Attribute()
size = model.Attribute(readonly=True)
uploaded_at = model.Attribute(readonly=True)
update_source = model.Attribute(readonly=True)
@property
def api(self):
return self.client.api.images[self.fingerprint]
@classmethod
def exists(cls, client, fingerprint, alias=False):
"""Determine whether an image exists.
architecture = model.Attribute()
config = model.Attribute()
created_at = model.Attribute()
devices = model.Attribute()
ephemeral = model.Attribute()
expanded_config = model.Attribute()
expanded_devices = model.Attribute()
name = model.Attribute(readonly=True)
description = model.Attribute()
profiles = model.Attribute()
status = model.Attribute(readonly=True)
last_used_at = model.Attribute(readonly=True)
location = model.Attribute(readonly=True)
status_code = model.Attribute(readonly=True)
stateful = model.Attribute(readonly=True)
snapshots = model.Manager()
files = model.Manager()
@property
def api(self):
return self.client.api.containers[self.name]
class FilesManager(object):
"""A pseudo-manager for namespacing file operations."""
def __init__(self, client, container):
self._client = client
self._container = container
def put(self, filepath, data, mode=None, uid=None, gid=None):
# under the License.
from pylxd.models import _model as model
class StoragePool(model.Model):
"""A LXD storage_pool.
This corresponds to the LXD endpoint at
/1.0/storage-pools
"""
name = model.Attribute()
driver = model.Attribute()
description = model.Attribute()
used_by = model.Attribute()
config = model.Attribute()
managed = model.Attribute()
@classmethod
def get(cls, client, name):
"""Get a storage_pool by name."""
response = client.api.storage_pools[name].get()
storage_pool = cls(client, **response.json()['metadata'])
return storage_pool
@classmethod
def all(cls, client):
"""Get all storage_pools."""
response = client.api.storage_pools.get()
storage_pools = []
for url in response.json()['metadata']:
@classmethod
def get(cls, client, *args):
"""Get cluster details"""
print(args)
response = client.api.cluster.get()
print(response.json())
container = cls(client, **response.json()['metadata'])
return container
class ClusterMember(model.Model):
"""A LXD cluster member."""
url = model.Attribute(readonly=True)
database = model.Attribute(readonly=True)
server_name = model.Attribute(readonly=True)
status = model.Attribute(readonly=True)
message = model.Attribute(readonly=True)
cluster = model.Parent()
@classmethod
def get(cls, client, server_name):
"""Get a cluster member by name."""
response = client.api.cluster.members[server_name].get()
return cls(client, **response.json()['metadata'])
@classmethod
def all(cls, client, *args):
"""Get all cluster members."""
class Container(model.Model):
"""An LXD Container.
This class is not intended to be used directly, but rather to be used
via `Client.containers.create`.
"""
architecture = model.Attribute()
config = model.Attribute()
created_at = model.Attribute()
devices = model.Attribute()
ephemeral = model.Attribute()
expanded_config = model.Attribute()
expanded_devices = model.Attribute()
name = model.Attribute(readonly=True)
description = model.Attribute()
profiles = model.Attribute()
status = model.Attribute(readonly=True)
last_used_at = model.Attribute(readonly=True)
location = model.Attribute(readonly=True)
status_code = model.Attribute(readonly=True)
stateful = model.Attribute(readonly=True)
snapshots = model.Manager()
files = model.Manager()
@property
def api(self):
return self.client.api.containers[self.name]
"""Get cluster details"""
print(args)
response = client.api.cluster.get()
print(response.json())
container = cls(client, **response.json()['metadata'])
return container
class ClusterMember(model.Model):
"""A LXD cluster member."""
url = model.Attribute(readonly=True)
database = model.Attribute(readonly=True)
server_name = model.Attribute(readonly=True)
status = model.Attribute(readonly=True)
message = model.Attribute(readonly=True)
cluster = model.Parent()
@classmethod
def get(cls, client, server_name):
"""Get a cluster member by name."""
response = client.api.cluster.members[server_name].get()
return cls(client, **response.json()['metadata'])
@classmethod
def all(cls, client, *args):
"""Get all cluster members."""
response = client.api.cluster.members.get()
nodes = []
def handshake_ok(self):
if self.payload:
if hasattr(self.payload, "read"):
self.send(
(self._smart_encode(line) for line in self.payload),
binary=True
)
else:
self.send(self._smart_encode(self.payload), binary=True)
self.send(b"", binary=False)
class Snapshot(model.Model):
"""A container snapshot."""
name = model.Attribute()
created_at = model.Attribute()
stateful = model.Attribute()
container = model.Parent()
@property
def api(self):
return self.client.api.containers[
self.container.name].snapshots[self.name]
@classmethod
def get(cls, client, container, name):
response = client.api.containers[
container.name].snapshots[name].get()
snapshot = cls(
response.json()['operation'])
return response.json()['operation']
class Image(model.Model):
"""A LXD Image."""
aliases = model.Attribute(readonly=True)
auto_update = model.Attribute(optional=True)
architecture = model.Attribute(readonly=True)
cached = model.Attribute(readonly=True)
created_at = model.Attribute(readonly=True)
expires_at = model.Attribute(readonly=True)
filename = model.Attribute(readonly=True)
fingerprint = model.Attribute(readonly=True)
last_used_at = model.Attribute(readonly=True)
properties = model.Attribute()
public = model.Attribute()
size = model.Attribute(readonly=True)
uploaded_at = model.Attribute(readonly=True)
update_source = model.Attribute(readonly=True)
@property
def api(self):
return self.client.api.images[self.fingerprint]
@classmethod
def exists(cls, client, fingerprint, alias=False):
"""Determine whether an image exists.
If `alias` is True, look up the image by its alias,
rather than its fingerprint.
"""