Secure your code as it's written. Use Snyk Code to scan source code in minutes - no build needed - and fix issues immediately.
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# 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()
@property
def api(self):
return self.client.api.cluster
@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'])
class ContainerState(object):
"""A simple object for representing container state."""
def __init__(self, **kwargs):
for key, value in six.iteritems(kwargs):
setattr(self, key, value)
_ContainerExecuteResult = collections.namedtuple(
'ContainerExecuteResult',
['exit_code', 'stdout', 'stderr'])
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()
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from pylxd.models import _model as model
from pylxd import managers
class Cluster(model.Model):
"""An LXD Cluster.
"""
server_name = model.Attribute()
enabled = model.Attribute()
member_config = model.Attribute()
members = model.Manager()
def __init__(self, *args, **kwargs):
super(Cluster, self).__init__(*args, **kwargs)
self.members = managers.ClusterMemberManager(self.client, self)
@property
def api(self):
return self.client.api.cluster
return msg.encode(self.encoding)
return msg
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[
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import json
from pylxd.exceptions import LXDAPIExtensionNotAvailable
from pylxd.models import _model as model
class Network(model.Model):
"""Model representing a LXD network."""
name = model.Attribute()
description = model.Attribute()
type = model.Attribute()
config = model.Attribute()
managed = model.Attribute(readonly=True)
used_by = model.Attribute(readonly=True)
@classmethod
def exists(cls, client, name):
"""
Determine whether network with provided name exists.
:param client: client instance
:type client: :class:`~pylxd.client.Client`
:param name: name of the network
from pylxd.models import _model as model
def _image_create_from_config(client, config, wait=False):
""" Create an image from the given configuration.
See: https://github.com/lxc/lxd/blob/master/doc/rest-api.md#post-6
"""
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)
def __setattr__(self, name, value):
if name in self.__attributes__:
attribute = self.__attributes__[name]
if attribute.validator is not None:
if attribute.validator is not type(value):
value = attribute.validator(value)
self.__dirty__.add(name)
return super(Model, self).__setattr__(name, value)
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
import binascii
from cryptography import x509
from cryptography.hazmat.backends import default_backend
from cryptography.hazmat.primitives import hashes
from cryptography.hazmat.primitives.serialization import Encoding
from pylxd.models import _model as model
class Certificate(model.Model):
"""A LXD certificate."""
certificate = model.Attribute()
fingerprint = model.Attribute()
type = model.Attribute()
name = model.Attribute()
@classmethod
def get(cls, client, fingerprint):
"""Get a certificate by fingerprint."""
response = client.api.certificates[fingerprint].get()
return cls(client, **response.json()['metadata'])
@classmethod
def all(cls, client):
#
# Licensed under the Apache License, Version 2.0 (the "License"); you may
# not use this file except in compliance with the License. You may obtain
# a copy of the License at
#
# http://www.apache.org/licenses/LICENSE-2.0
#
# Unless required by applicable law or agreed to in writing, software
# distributed under the License is distributed on an "AS IS" BASIS, WITHOUT
# WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. See the
# License for the specific language governing permissions and limitations
# under the License.
from pylxd.models import _model as model
class Profile(model.Model):
"""A LXD profile."""
config = model.Attribute()
description = model.Attribute()
devices = model.Attribute()
name = model.Attribute(readonly=True)
used_by = model.Attribute(readonly=True)
@classmethod
def exists(cls, client, name):
"""Determine whether a profile exists."""
try:
client.profiles.get(name)
return True
except cls.NotFound:
return False