/usr/lib/python3/dist-packages/etcd3gw/lease.py is in python3-etcd3gw 0.2.1-1.
This file is owned by root:root, with mode 0o644.
The actual contents of the file can be viewed below.
1 2 3 4 5 6 7 8 9 10 11 12 13 14 15 16 17 18 19 20 21 22 23 24 25 26 27 28 29 30 31 32 33 34 35 36 37 38 39 40 41 42 43 44 45 46 47 48 49 50 51 52 53 54 55 56 57 58 59 60 61 62 63 64 65 66 67 68 69 70 71 72 73 74 75 | # 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 etcd3gw.utils import _decode
class Lease(object):
def __init__(self, id, client=None):
"""Lease object for expiring keys
:param id:
:param client:
"""
self.id = id
self.client = client
def revoke(self):
"""LeaseRevoke revokes a lease.
All keys attached to the lease will expire and be deleted.
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
:return:
"""
self.client.post(self.client.get_url("/kv/lease/revoke"),
json={"ID": self.id})
return True
def ttl(self):
"""LeaseTimeToLive retrieves lease information.
This method makes a synchronous HTTP request by default. To make an
asynchronous HTTP request, please define a `callback` function
to be invoked when receiving the response.
:return:
"""
result = self.client.post(self.client.get_url("/kv/lease/timetolive"),
json={"ID": self.id})
return int(result['TTL'])
def refresh(self):
"""LeaseKeepAlive keeps the lease alive
By streaming keep alive requests from the client to the server and
streaming keep alive responses from the server to the client.
This method makes a synchronous HTTP request by default.
:return:
"""
result = self.client.post(self.client.get_url("/lease/keepalive"),
json={"ID": self.id})
return int(result['result']['TTL'])
def keys(self):
"""Get the keys associated with this lease.
:return:
"""
result = self.client.post(self.client.get_url("/kv/lease/timetolive"),
json={"ID": self.id,
"keys": True})
keys = result['keys'] if 'keys' in result else []
return [_decode(key) for key in keys]
|