Merge pull request #31 from docker/staging

1.0.10

Author: Feng Honglin

.gitignore

@@ -58,4 +58,7 @@ target/
 # IDE
 .idea/
 
+# vim
+*.swp
+
 venv/

README.md

@@ -29,6 +29,20 @@ The authentication can be configured in the following ways:
 		export DOCKERCLOUD_USER=username
 		export DOCKERCLOUD_APIKEY=apikey
 
+## Optional parameters
+
+You may set the reconnection interval (Integer, in seconds) using the variable DOCKERCLOUD_RECONNECTION_INTERVAL:
+
+		export DOCKERCLOUD_RECONNECTION_INTERVAL=240
+
+Session uses a socket that may be closed by some peer. To prevent the "Read timed out" issue you should use this option.
+
+Possible values:
+
+* `-1` (by default) means no reconnect (as usually it works)
+* `0` means reconnect on each request
+* any positive value means that the connection will be reopened if the time diff between last 2 requests is more than that value
+
 ## Namespace
 
 To support teams and orgs, you can specify the namespace in the following ways:

dockercloud/__init__.py

@@ -25,7 +25,7 @@
 from dockercloud.api.events import Events
 from dockercloud.api.nodeaz import AZ
 
-__version__ = '1.0.9'
+__version__ = '1.0.10'
 
 dockercloud_auth = os.environ.get('DOCKERCLOUD_AUTH')
 basic_auth = auth.load_from_file("~/.docker/config.json")
@@ -40,8 +40,14 @@
 
 namespace = os.environ.get('DOCKERCLOUD_NAMESPACE')
 
+# in seconds, if the connection is inactive more than that value it will be recreated
+reconnection_interval = int(os.environ.get('DOCKERCLOUD_RECONNECTION_INTERVAL', '-1'))
+
 user_agent = None
 
+# in seconds, make the api call timeout after X seconds, None usually is 15 mins
+api_timeout = None
+
 logging.basicConfig()
 logger = logging.getLogger("python-dockercloud")
 

dockercloud/api/action.py

@@ -6,7 +6,7 @@
 class Action(Immutable):
     subsystem = 'audit'
     endpoint = "/action"
-    namespaced = False
+    is_namespaced = False
 
     @classmethod
     def _pk_key(cls):

dockercloud/api/auth.py

@@ -12,6 +12,7 @@
 
 HUB_INDEX = "https://index.docker.io/v1/"
 
+
 def authenticate(username, password):
     verify_credential(username, password)
     dockercloud.basic_auth = base64.b64encode("%s:%s" % (username, password))
@@ -55,7 +56,8 @@ def load_from_file(f="~/.docker/config.json"):
             p = subprocess.Popen([cmd, 'get'], stdout=subprocess.PIPE, stdin=subprocess.PIPE, stderr=subprocess.STDOUT)
             out = p.communicate(input=HUB_INDEX)[0]
         except:
-            raise dockercloud.AuthError('error getting credentials - err: exec: "%s": executable file not found in $PATH, out: ``' % cmd)
+            raise dockercloud.AuthError(
+                'error getting credentials - err: exec: "%s": executable file not found in $PATH, out: ``' % cmd)
 
         try:
             credential = json.loads(out)

dockercloud/api/base.py

@@ -16,18 +16,19 @@
 class BasicObject(object):
     _api_version = 'v1'
 
-    def __init__(self, **kwargs):
-        pass
-
 
 class Restful(BasicObject):
-    _detail_uri = None
-    namespaced = True
+    is_namespaced = True
 
-    def __init__(self, **kwargs):
+    def __init__(self, namespace="", **kwargs):
         """Simply reflect all the values in kwargs"""
         for k, v in list(kwargs.items()):
             setattr(self, k, v)
+        if self.is_namespaced and namespace:
+            self._namespace = namespace
+        else:
+            self._namespace = dockercloud.namespace
+        self._resource_uri = ""
 
     def __addchanges__(self, name):
         changed_attrs = self.__getchanges__()
@@ -38,7 +39,7 @@ def __addchanges__(self, name):
     def __setattr__(self, name, value):
         """Keeps track of what attributes have been set"""
         current_value = getattr(self, name, None)
-        if value != current_value:
+        if value != current_value and not name.startswith("_"):
             self.__addchanges__(name)
         super(Restful, self).__setattr__(name, value)
 
@@ -53,17 +54,10 @@ def __setchanges__(self, val):
 
     def _loaddict(self, dict):
         """Internal. Sets the model attributes to the dictionary values passed"""
-        endpoint = getattr(self, 'endpoint', None)
-        subsystem = getattr(self, 'subsystem', None)
-        assert endpoint, "Endpoint not specified for %s" % self.__class__.__name__
-        assert subsystem, "Subsystem not specified for %s" % self.__class__.__name__
         for k, v in list(dict.items()):
             setattr(self, k, v)
-        if self.namespaced and dockercloud.namespace:
-            self._detail_uri = "/".join(["api", subsystem, self._api_version, dockercloud.namespace,
-                                         endpoint.strip("/"), self.pk])
-        else:
-            self._detail_uri = "/".join(["api", subsystem, self._api_version, endpoint.strip("/"), self.pk])
+
+        self._resource_uri = getattr(self, "resource_uri", None)
         self.__setchanges__([])
 
     @property
@@ -93,9 +87,9 @@ def is_dirty(self):
     def _perform_action(self, action, params=None, data={}):
         """Internal. Performs the specified action on the object remotely"""
         success = False
-        if not self._detail_uri:
+        if not self._resource_uri:
             raise ApiError("You must save the object before performing this operation")
-        path = "/".join([self._detail_uri.rstrip("/"), action.lstrip("/")])
+        path = "/".join([self._resource_uri.rstrip("/"), action.lstrip("/")])
         json = send_request("POST", path, params=params, data=data)
         if json:
             self._loaddict(json)
@@ -104,9 +98,9 @@ def _perform_action(self, action, params=None, data={}):
 
     def _expand_attribute(self, attribute):
         """Internal. Expands the given attribute from remote information"""
-        if not self._detail_uri:
+        if not self._resource_uri:
             raise ApiError("You must save the object before performing this operation")
-        path = "/".join([self._detail_uri, attribute])
+        path = "/".join([self._resource_uri, attribute])
         json = send_request("GET", path)
         if json:
             return json[attribute]
@@ -125,39 +119,43 @@ def get_all_attributes(self):
 
 class Immutable(Restful):
     @classmethod
-    def fetch(cls, pk):
-        instance = None
+    def fetch(cls, pk, namespace=""):
         endpoint = getattr(cls, 'endpoint', None)
         subsystem = getattr(cls, 'subsystem', None)
         assert endpoint, "Endpoint not specified for %s" % cls.__name__
         assert subsystem, "Subsystem not specified for %s" % cls.__name__
-        if cls.namespaced and dockercloud.namespace:
-            detail_uri = "/".join(["api", subsystem, cls._api_version, dockercloud.namespace, endpoint.strip("/"), pk])
+
+        if not namespace:
+            namespace = dockercloud.namespace
+        if cls.is_namespaced and namespace:
+            resource_uri = "/".join(["api", subsystem, cls._api_version, namespace, endpoint.strip("/"), pk])
         else:
-            detail_uri = "/".join(["api", subsystem, cls._api_version, endpoint.strip("/"), pk])
-        json = send_request('GET', detail_uri)
+            resource_uri = "/".join(["api", subsystem, cls._api_version, endpoint.strip("/"), pk])
+        json = send_request('GET', resource_uri)
         if json:
             instance = cls()
             instance._loaddict(json)
         return instance
 
     @classmethod
-    def list(cls, limit=None, **kwargs):
+    def list(cls, limit=None, namespace="", **kwargs):
         restful = []
         endpoint = getattr(cls, 'endpoint', None)
         subsystem = getattr(cls, 'subsystem', None)
         assert endpoint, "Endpoint not specified for %s" % cls.__name__
         assert subsystem, "Subsystem not specified for %s" % cls.__name__
 
-        if cls.namespaced and dockercloud.namespace:
-            detail_uri = "/".join(["api", subsystem, cls._api_version, dockercloud.namespace, endpoint.strip("/")])
+        if not namespace:
+            namespace = dockercloud.namespace
+        if cls.is_namespaced and namespace:
+            resource_uri = "/".join(["api", subsystem, cls._api_version, namespace, endpoint.strip("/")])
         else:
-            detail_uri = "/".join(["api", subsystem, cls._api_version, endpoint.strip("/")])
+            resource_uri = "/".join(["api", subsystem, cls._api_version, endpoint.strip("/")])
         objects = []
         while True:
             if limit and len(objects) >= limit:
                 break
-            json = send_request('GET', detail_uri, params=kwargs)
+            json = send_request('GET', resource_uri, params=kwargs)
             objs = json.get('objects', [])
             meta = json.get('meta', {})
             next_url = meta.get('next', '')
@@ -182,10 +180,10 @@ def refresh(self, force=False):
         if self.is_dirty and not force:
             # We have local non-committed changes - rejecting the refresh
             success = False
-        elif not self._detail_uri:
+        elif not self._resource_uri:
             raise ApiError("You must save the object before performing this operation")
         else:
-            json = send_request("GET", self._detail_uri)
+            json = send_request("GET", self._resource_uri)
             if json:
                 self._loaddict(json)
                 success = True
@@ -202,16 +200,17 @@ def create(cls, **kwargs):
         return cls(**kwargs)
 
     def delete(self):
-        if not self._detail_uri:
+        if not self._resource_uri:
             raise ApiError("You must save the object before performing this operation")
         action = "DELETE"
-        url = self._detail_uri
+        url = self._resource_uri
         json = send_request(action, url)
         if json:
             self._loaddict(json)
+            self._resource_uri = None
         else:
             # Object deleted successfully and nothing came back - deleting PK reference.
-            self._detail_uri = None
+            self._resource_uri = None
             # setattr(self, self._pk_key(), None) -- doesn't work
             self.__setchanges__([])
         return True
@@ -228,15 +227,15 @@ def save(self):
             assert endpoint, "Endpoint not specified for %s" % self.__class__.__name__
             assert subsystem, "Subsystem not specified for %s" % self.__class__.__name__
             # Figure out whether we should do a create or update
-            if not self._detail_uri:
+            if not self._resource_uri:
                 action = "POST"
-                if cls.namespaced and dockercloud.namespace:
-                    path = "/".join(["api", subsystem, self._api_version, dockercloud.namespace, endpoint.lstrip("/")])
+                if cls.is_namespaced and self._namespace:
+                    path = "/".join(["api", subsystem, self._api_version, self._namespace, endpoint.lstrip("/")])
                 else:
                     path = "/".join(["api", subsystem, self._api_version, endpoint.lstrip("/")])
             else:
                 action = "PATCH"
-                path = self._detail_uri
+                path = self._resource_uri
             # Construct the necessary params
             params = {}
             for attr in self.__getchanges__():
@@ -322,13 +321,16 @@ def run_forever(self, *args, **kwargs):
 
 
 class StreamingLog(StreamingAPI):
-    def __init__(self, subsystem, resource, uuid, tail, follow):
+    def __init__(self, subsystem, resource, uuid, tail, follow, namespace=""):
         endpoint = "%s/%s/logs/?follow=%s" % (resource, uuid, str(follow).lower())
         if tail:
             endpoint = "%s&tail=%d" % (endpoint, tail)
-        if dockercloud.namespace:
+
+        if not namespace:
+            namespace = dockercloud.namespace
+        if namespace:
             url = "/".join([dockercloud.stream_host.rstrip("/"), "api", subsystem, self._api_version,
-                            dockercloud.namespace, endpoint.lstrip("/")])
+                            self._namespace, endpoint.lstrip("/")])
         else:
             url = "/".join([dockercloud.stream_host.rstrip("/"), "api", subsystem, self._api_version,
                             endpoint.lstrip("/")])
@@ -348,11 +350,13 @@ def run_forever(self, *args, **kwargs):
 
 
 class Exec(StreamingAPI):
-    def __init__(self, uuid, cmd='sh'):
+    def __init__(self, uuid, cmd='sh', namespace=""):
         endpoint = "container/%s/exec/?command=%s" % (uuid, urllib.quote_plus(cmd))
-        if dockercloud.namespace:
+        if not namespace:
+            namespace = dockercloud.namespace
+        if namespace:
             url = "/".join([dockercloud.stream_host.rstrip("/"), "api", "app", self._api_version,
-                            dockercloud.namespace, endpoint.lstrip("/")])
+                            namespace, endpoint.lstrip("/")])
         else:
             url = "/".join([dockercloud.stream_host.rstrip("/"), "api", "app", self._api_version, endpoint.lstrip("/")])
         super(self.__class__, self).__init__(url)

dockercloud/api/events.py

@@ -2,6 +2,7 @@
 
 import json
 import logging
+import signal
 
 import websocket
 
@@ -13,11 +14,14 @@
 
 
 class Events(StreamingAPI):
-    def __init__(self):
+    def __init__(self, namespace=""):
         endpoint = "events"
-        if dockercloud.namespace:
+
+        if not namespace:
+            namespace = dockercloud.namespace
+        if namespace:
             url = "/".join([dockercloud.stream_host.rstrip("/"), "api", "audit", self._api_version,
-                            dockercloud.namespace, endpoint.lstrip("/")])
+                            namespace, endpoint.lstrip("/")])
         else:
             url = "/".join([dockercloud.stream_host.rstrip("/"), "api", "audit", self._api_version,
                             endpoint.lstrip("/")])
@@ -41,15 +45,22 @@ def _on_error(self, ws, e):
 
         super(self.__class__, self)._on_error(ws, e)
 
+    def _on_stop(self, signal, frame):
+        self.ws.close()
+        self.run_forever_flag = not self.run_forever_flag
+
     def run_forever(self, *args, **kwargs):
-        while True:
+
+        self.run_forever_flag = True
+        while self.run_forever_flag:
             if self.auth_error:
                 self.auth_error = False
                 raise AuthError("Not Authorized")
 
-            ws = websocket.WebSocketApp(self.url, header=self.header,
-                                        on_open=self._on_open,
-                                        on_message=self._on_message,
-                                        on_error=self._on_error,
-                                        on_close=self._on_close)
-            ws.run_forever(ping_interval=10, ping_timeout=5, *args, **kwargs)
+            self.ws = websocket.WebSocketApp(self.url, header=self.header,
+                                             on_open=self._on_open,
+                                             on_message=self._on_message,
+                                             on_error=self._on_error,
+                                             on_close=self._on_close)
+            signal.signal(signal.SIGINT, self._on_stop)
+            self.ws.run_forever(ping_interval=10, ping_timeout=5, *args, **kwargs)

dockercloud/api/http.py

@@ -1,6 +1,7 @@
 from __future__ import absolute_import
 
 import logging
+import time
 
 from requests import Request, Session
 from requests import utils
@@ -12,9 +13,15 @@
 logger = logging.getLogger("python-dockercloud")
 
 global_session = Session()
+last_connection_time = time.time()
 
 
-def get_session():
+def get_session(time=time):
+    if (dockercloud.reconnection_interval >= 0):
+        global last_connection_time
+        if (time.time() - last_connection_time > dockercloud.reconnection_interval):
+            new_session()
+        last_connection_time = time.time()
     return global_session
 
 
@@ -55,7 +62,12 @@ def send_request(method, path, inject_header=True, **kwargs):
     # make the request
     req = s.prepare_request(request)
     logger.info("Prepared Request: %s, %s, %s, %s" % (req.method, req.url, req.headers, kwargs))
-    response = s.send(req, **kw_args)
+
+    if dockercloud.api_timeout:
+        response = s.send(req, timeout=dockercloud.api_timeout, **kw_args)
+    else:
+        response = s.send(req, **kw_args)
+
     status_code = getattr(response, 'status_code', None)
     logger.info("Response: Status %s, %s, %s" % (str(status_code), response.headers, response.text))