Skip to content

R
rota
Commit bb846822 authored by 刘丙寅's avatar 刘丙寅
Browse files
Options

修改

parent 3bf81dea
Hide whitespace changes
Inline Side-by-side
Showing with 3568 additions and 47 deletions
.idea/workspace.xml
View file @ bb846822
...@@ -3,7 +3,7 @@ ...@@ -3,7 +3,7 @@
<component name="ChangeListManager"> <component name="ChangeListManager">
<list default="true" id="a188f8e8-7383-4735-a16a-e71d8f97fff3" name="Default Changelist" comment=""> <list default="true" id="a188f8e8-7383-4735-a16a-e71d8f97fff3" name="Default Changelist" comment="">
<change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" /> <change beforePath="$PROJECT_DIR$/.idea/workspace.xml" beforeDir="false" afterPath="$PROJECT_DIR$/.idea/workspace.xml" afterDir="false" />
<change beforePath="$PROJECT_DIR$/peizhi.py" beforeDir="false" afterPath="$PROJECT_DIR$/peizhi.py" afterDir="false" /> <change beforePath="$PROJECT_DIR$/rota_gm.py" beforeDir="false" afterPath="$PROJECT_DIR$/rota_gm.py" afterDir="false" />
</list> </list>
<option name="EXCLUDED_CONVERTED_TO_IGNORED" value="true" /> <option name="EXCLUDED_CONVERTED_TO_IGNORED" value="true" />
<option name="SHOW_DIALOG" value="false" /> <option name="SHOW_DIALOG" value="false" />
...@@ -21,11 +21,10 @@ ...@@ -21,11 +21,10 @@
<file pinned="false" current-in-tab="false"> <file pinned="false" current-in-tab="false">
<entry file="file://$PROJECT_DIR$/rota_gm_new.py"> <entry file="file://$PROJECT_DIR$/rota_gm_new.py">
<provider selected="true" editor-type-id="text-editor"> <provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="299"> <state relative-caret-position="35581">
<caret line="13" column="49" lean-forward="true" selection-start-line="13" selection-start-column="49" selection-end-line="13" selection-end-column="49" /> <caret line="1550" column="19" selection-start-line="1550" selection-start-column="19" selection-end-line="1550" selection-end-column="19" />
<folding> <folding>
<element signature="e#0#15#0" expanded="true" /> <element signature="e#0#15#0" expanded="true" />
<element signature="e#694#134799#0" />
</folding> </folding>
</state> </state>
</provider> </provider>
...@@ -34,8 +33,29 @@ ...@@ -34,8 +33,29 @@
<file pinned="false" current-in-tab="true"> <file pinned="false" current-in-tab="true">
<entry file="file://$PROJECT_DIR$/peizhi.py"> <entry file="file://$PROJECT_DIR$/peizhi.py">
<provider selected="true" editor-type-id="text-editor"> <provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="253"> <state relative-caret-position="138">
<caret line="11" lean-forward="true" selection-start-line="11" selection-end-line="11" /> <caret line="6" column="21" lean-forward="true" selection-start-line="6" selection-start-column="21" selection-end-line="6" selection-end-column="21" />
</state>
</provider>
</entry>
</file>
<file pinned="false" current-in-tab="false">
<entry file="file://$PROJECT_DIR$/rota_gm.py">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="805">
<caret line="37" selection-start-line="37" selection-end-line="37" />
</state>
</provider>
</entry>
</file>
<file pinned="false" current-in-tab="false">
<entry file="file://$PROJECT_DIR$/demo.py">
<provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="1633">
<caret line="73" column="26" selection-start-line="73" selection-start-column="26" selection-end-line="73" selection-end-column="26" />
<folding>
<element signature="e#0#13#0" expanded="true" />
</folding>
</state> </state>
</provider> </provider>
</entry> </entry>
...@@ -67,16 +87,16 @@ ...@@ -67,16 +87,16 @@
<component name="IdeDocumentHistory"> <component name="IdeDocumentHistory">
<option name="CHANGED_PATHS"> <option name="CHANGED_PATHS">
<list> <list>
<option value="$PROJECT_DIR$/rota_gm.py" />
<option value="$PROJECT_DIR$/demo.py" /> <option value="$PROJECT_DIR$/demo.py" />
<option value="$PROJECT_DIR$/rota_gm_new.py" /> <option value="$PROJECT_DIR$/rota_gm_new.py" />
<option value="$PROJECT_DIR$/rota_gm.py" />
<option value="$PROJECT_DIR$/peizhi.py" /> <option value="$PROJECT_DIR$/peizhi.py" />
</list> </list>
</option> </option>
</component> </component>
<component name="ProjectFrameBounds" extendedState="6"> <component name="ProjectFrameBounds">
<option name="x" value="-1920" /> <option name="x" value="-1920" />
<option name="y" value="-157" /> <option name="y" value="-52" />
<option name="width" value="1920" /> <option name="width" value="1920" />
<option name="height" value="1057" /> <option name="height" value="1057" />
</component> </component>
...@@ -86,6 +106,7 @@ ...@@ -86,6 +106,7 @@
<foldersAlwaysOnTop value="true" /> <foldersAlwaysOnTop value="true" />
</navigator> </navigator>
<panes> <panes>
<pane id="Scope" />
<pane id="ProjectPane"> <pane id="ProjectPane">
<subPane> <subPane>
<expand> <expand>
...@@ -97,7 +118,6 @@ ...@@ -97,7 +118,6 @@
<select /> <select />
</subPane> </subPane>
</pane> </pane>
<pane id="Scope" />
</panes> </panes>
</component> </component>
<component name="PropertiesComponent"> <component name="PropertiesComponent">
...@@ -212,7 +232,15 @@ ...@@ -212,7 +232,15 @@
<workItem from="1623760622650" duration="150000" /> <workItem from="1623760622650" duration="150000" />
<workItem from="1624005728824" duration="70000" /> <workItem from="1624005728824" duration="70000" />
<workItem from="1624413811074" duration="2649000" /> <workItem from="1624413811074" duration="2649000" />
<workItem from="1624589403519" duration="1162000" /> <workItem from="1624589403519" duration="1547000" />
<workItem from="1625222905796" duration="120000" />
<workItem from="1628842817789" duration="609000" />
<workItem from="1630029757582" duration="834000" />
<workItem from="1630051537497" duration="761000" />
<workItem from="1630656566556" duration="395000" />
<workItem from="1630657105124" duration="68000" />
<workItem from="1631873368095" duration="216000" />
<workItem from="1632449641846" duration="335000" />
</task> </task>
<task id="LOCAL-00001" summary="修改文案和时间"> <task id="LOCAL-00001" summary="修改文案和时间">
<created>1619687262591</created> <created>1619687262591</created>
...@@ -284,11 +312,60 @@ ...@@ -284,11 +312,60 @@
<option name="project" value="LOCAL" /> <option name="project" value="LOCAL" />
<updated>1624590586340</updated> <updated>1624590586340</updated>
</task> </task>
<option name="localTasksCounter" value="11" /> <task id="LOCAL-00011" summary="优化代码1">
<created>1624590857333</created>
<option name="number" value="00011" />
<option name="presentableId" value="LOCAL-00011" />
<option name="project" value="LOCAL" />
<updated>1624590857333</updated>
</task>
<task id="LOCAL-00012" summary="接触钉钉报警注释">
<created>1624591041604</created>
<option name="number" value="00012" />
<option name="presentableId" value="LOCAL-00012" />
<option name="project" value="LOCAL" />
<updated>1624591041604</updated>
</task>
<task id="LOCAL-00013" summary="值班">
<created>1630030076173</created>
<option name="number" value="00013" />
<option name="presentableId" value="LOCAL-00013" />
<option name="project" value="LOCAL" />
<updated>1630030076173</updated>
</task>
<task id="LOCAL-00014" summary="值班1">
<created>1630051778079</created>
<option name="number" value="00014" />
<option name="presentableId" value="LOCAL-00014" />
<option name="project" value="LOCAL" />
<updated>1630051778079</updated>
</task>
<task id="LOCAL-00015" summary="离职人员删除">
<created>1630656822327</created>
<option name="number" value="00015" />
<option name="presentableId" value="LOCAL-00015" />
<option name="project" value="LOCAL" />
<updated>1630656822327</updated>
</task>
<task id="LOCAL-00016" summary="离职人员删除">
<created>1630656914711</created>
<option name="number" value="00016" />
<option name="presentableId" value="LOCAL-00016" />
<option name="project" value="LOCAL" />
<updated>1630656914711</updated>
</task>
<task id="LOCAL-00017" summary="修改">
<created>1630657160233</created>
<option name="number" value="00017" />
<option name="presentableId" value="LOCAL-00017" />
<option name="project" value="LOCAL" />
<updated>1630657160233</updated>
</task>
<option name="localTasksCounter" value="18" />
<servers /> <servers />
</component> </component>
<component name="TimeTrackingManager"> <component name="TimeTrackingManager">
<option name="totallyTimeSpent" value="43370000" /> <option name="totallyTimeSpent" value="47093000" />
</component> </component>
<component name="TodoView"> <component name="TodoView">
<todo-panel id="selected-file"> <todo-panel id="selected-file">
...@@ -300,10 +377,10 @@ ...@@ -300,10 +377,10 @@
</todo-panel> </todo-panel>
</component> </component>
<component name="ToolWindowManager"> <component name="ToolWindowManager">
<frame x="-1920" y="-157" width="1920" height="1057" extended-state="6" /> <frame x="-1920" y="-52" width="1920" height="1057" extended-state="0" />
<editor active="true" /> <editor active="true" />
<layout> <layout>
<window_info content_ui="combo" id="Project" order="0" visible="true" weight="0.10596379" /> <window_info active="true" content_ui="combo" id="Project" order="0" visible="true" weight="0.08413205" />
<window_info id="Structure" order="1" side_tool="true" weight="0.25" /> <window_info id="Structure" order="1" side_tool="true" weight="0.25" />
<window_info id="Favorites" order="2" side_tool="true" /> <window_info id="Favorites" order="2" side_tool="true" />
<window_info anchor="bottom" id="Message" order="0" /> <window_info anchor="bottom" id="Message" order="0" />
...@@ -339,7 +416,13 @@ ...@@ -339,7 +416,13 @@
<MESSAGE value="jenkins调试1" /> <MESSAGE value="jenkins调试1" />
<MESSAGE value="更新代码" /> <MESSAGE value="更新代码" />
<MESSAGE value="优化代码" /> <MESSAGE value="优化代码" />
<option name="LAST_COMMIT_MESSAGE" value="优化代码" /> <MESSAGE value="优化代码1" />
<MESSAGE value="接触钉钉报警注释" />
<MESSAGE value="值班" />
<MESSAGE value="值班1" />
<MESSAGE value="离职人员删除" />
<MESSAGE value="修改" />
<option name="LAST_COMMIT_MESSAGE" value="修改" />
</component> </component>
<component name="editorHistoryManager"> <component name="editorHistoryManager">
<entry file="file://$PROJECT_DIR$/venv/lib/python3.7/site-packages/requests/adapters.py"> <entry file="file://$PROJECT_DIR$/venv/lib/python3.7/site-packages/requests/adapters.py">
...@@ -536,38 +619,37 @@ ...@@ -536,38 +619,37 @@
</state> </state>
</provider> </provider>
</entry> </entry>
<entry file="file://$PROJECT_DIR$/demo.py"> <entry file="file://$PROJECT_DIR$/rota_gm_new.py">
<provider selected="true" editor-type-id="text-editor"> <provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="1114"> <state relative-caret-position="35581">
<caret line="73" column="26" lean-forward="true" selection-start-line="73" selection-start-column="26" selection-end-line="73" selection-end-column="26" /> <caret line="1550" column="19" selection-start-line="1550" selection-start-column="19" selection-end-line="1550" selection-end-column="19" />
<folding> <folding>
<element signature="e#0#13#0" expanded="true" /> <element signature="e#0#15#0" expanded="true" />
</folding> </folding>
</state> </state>
</provider> </provider>
</entry> </entry>
<entry file="file://$PROJECT_DIR$/rota_gm.py"> <entry file="file://$PROJECT_DIR$/rota_gm.py">
<provider selected="true" editor-type-id="text-editor"> <provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="618"> <state relative-caret-position="805">
<caret line="46" column="25" lean-forward="true" selection-start-line="46" selection-start-column="25" selection-end-line="46" selection-end-column="25" /> <caret line="37" selection-start-line="37" selection-end-line="37" />
</state> </state>
</provider> </provider>
</entry> </entry>
<entry file="file://$PROJECT_DIR$/rota_gm_new.py"> <entry file="file://$PROJECT_DIR$/demo.py">
<provider selected="true" editor-type-id="text-editor"> <provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="299"> <state relative-caret-position="1633">
<caret line="13" column="49" lean-forward="true" selection-start-line="13" selection-start-column="49" selection-end-line="13" selection-end-column="49" /> <caret line="73" column="26" selection-start-line="73" selection-start-column="26" selection-end-line="73" selection-end-column="26" />
<folding> <folding>
<element signature="e#0#15#0" expanded="true" /> <element signature="e#0#13#0" expanded="true" />
<element signature="e#694#134799#0" />
</folding> </folding>
</state> </state>
</provider> </provider>
</entry> </entry>
<entry file="file://$PROJECT_DIR$/peizhi.py"> <entry file="file://$PROJECT_DIR$/peizhi.py">
<provider selected="true" editor-type-id="text-editor"> <provider selected="true" editor-type-id="text-editor">
<state relative-caret-position="253"> <state relative-caret-position="138">
<caret line="11" lean-forward="true" selection-start-line="11" selection-end-line="11" /> <caret line="6" column="21" lean-forward="true" selection-start-line="6" selection-start-column="21" selection-end-line="6" selection-end-column="21" />
</state> </state>
</provider> </provider>
</entry> </entry>
......
demo.py 0 → 100644
View file @ bb846822
import random
import redis,json
import requests
r = redis.Redis()
import peizhi
def test111():
list = r.lrange("zhiban",0,-1)
r.delete('zhiban')
print(list)
if not list:
list = ["张三","李四","王五","赵六","奉七","111"]
little_list = random.sample(list,2)
list.remove(little_list[0])
list.remove(little_list[1])
# little_list 是值班人员
r.lpush("zhiban", *list)
else:
little_list = random.sample(list, 2)
list.remove(little_list[0])
list.remove(little_list[1])
# little_list 是值班人员
if list == []:
pass
else:
r.lpush("zhiban", *list)
def demo222():
ll = ["丙寅","奉龙"]
phonenum = []
for i in ll:
phone = peizhi.Personnel_list_111.get(i)
phonenum.append(phone)
dd_url = "https://oapi.dingtalk.com/robot/send?access_token=b2f03936be10b15bd6395a2e567a4c41c0540a9f109c5046a4cdc3f85ac8e374"
text = "报警周末值班，请查看邮件排查问题"
json_text = {
"msgtype": "text",
"at": {
"atMobiles": phonenum,
# 变为true 就会@所有人
# "isAtAll": False
"isAtAll": False
},
"text": {
"content": text
}
}
requests.post(url=dd_url, json=json_text, verify=False)
def demo333():
ll = ["1","2"]
lll = ["1","2","3","5"]
lll.remove(ll)
print(lll)
if __name__ == '__main__':
demo333()
# 查出来明天休假开始 后面有多少天
# 获取到天数
# 根据天数 获取redis里面的人 如果人数不够 追加一个新列表
rota_gm.py
View file @ bb846822
...@@ -37,24 +37,24 @@ def test(): ...@@ -37,24 +37,24 @@ def test():
# 钉钉报警 # 钉钉报警
# dd_url = "https://oapi.dingtalk.com/robot/send?access_token=062ab7f98f47c85e6c200efd21ae5d532e4f1bbfd6782ae1268b9a1f4d3dc7b9" dd_url = "https://oapi.dingtalk.com/robot/send?access_token=062ab7f98f47c85e6c200efd21ae5d532e4f1bbfd6782ae1268b9a1f4d3dc7b9"
# text = "明日开始休息：本次值班人员为：%s 值班人员记得带 电脑，手机， vpn环境" % {message} text = "明日开始休息：本次值班人员为：%s 值班人员记得带 电脑，手机， vpn环境" % {message}
# json_text = { json_text = {
# "msgtype": "text", "msgtype": "text",
# "at": { "at": {
# "atMobiles": [ "atMobiles": [
#
# "all" "all"
# ], ],
# # 变为true 就会@所有人 # 变为true 就会@所有人
# "isAtAll": True "isAtAll": True
# }, },
# "text": { "text": {
# "content": text "content": text
# } }
# } }
#
# requests.post(url=dd_url, json=json_text, verify=False) requests.post(url=dd_url, json=json_text, verify=False)
......
venv/lib/python3.7/site-packages/redis-3.5.3.dist-info/LICENSE 0 → 100644
View file @ bb846822
Copyright (c) 2012 Andy McCurdy
Permission is hereby granted, free of charge, to any person
obtaining a copy of this software and associated documentation
files (the "Software"), to deal in the Software without
restriction, including without limitation the rights to use,
copy, modify, merge, publish, distribute, sublicense, and/or sell
copies of the Software, and to permit persons to whom the
Software is furnished to do so, subject to the following
conditions:
The above copyright notice and this permission notice shall be
included in all copies or substantial portions of the Software.
THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND,
EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES
OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND
NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT
HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY,
WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING
FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR
OTHER DEALINGS IN THE SOFTWARE.
venv/lib/python3.7/site-packages/redis-3.5.3.dist-info/METADATA 0 → 100644
View file @ bb846822
Metadata-Version: 2.1
Name: redis
Version: 3.5.3
Summary: Python client for Redis key-value store
Home-page: https://github.com/andymccurdy/redis-py
Author: Andy McCurdy
Author-email: sedrik@gmail.com
Maintainer: Andy McCurdy
Maintainer-email: sedrik@gmail.com
License: MIT
Keywords: Redis,key-value store
Platform: UNKNOWN
Classifier: Development Status :: 5 - Production/Stable
Classifier: Environment :: Console
Classifier: Intended Audience :: Developers
Classifier: License :: OSI Approved :: MIT License
Classifier: Operating System :: OS Independent
Classifier: Programming Language :: Python
Classifier: Programming Language :: Python :: 2
Classifier: Programming Language :: Python :: 2.7
Classifier: Programming Language :: Python :: 3
Classifier: Programming Language :: Python :: 3.5
Classifier: Programming Language :: Python :: 3.6
Classifier: Programming Language :: Python :: 3.7
Classifier: Programming Language :: Python :: 3.8
Classifier: Programming Language :: Python :: Implementation :: CPython
Classifier: Programming Language :: Python :: Implementation :: PyPy
Requires-Python: >=2.7, !=3.0.*, !=3.1.*, !=3.2.*, !=3.3.*, !=3.4.*
Provides-Extra: hiredis
Requires-Dist: hiredis (>=0.1.3) ; extra == 'hiredis'
redis-py
========
The Python interface to the Redis key-value store.
.. image:: https://secure.travis-ci.org/andymccurdy/redis-py.svg?branch=master
:target: https://travis-ci.org/andymccurdy/redis-py
.. image:: https://readthedocs.org/projects/redis-py/badge/?version=stable&style=flat
:target: https://redis-py.readthedocs.io/en/stable/
.. image:: https://badge.fury.io/py/redis.svg
:target: https://pypi.org/project/redis/
.. image:: https://codecov.io/gh/andymccurdy/redis-py/branch/master/graph/badge.svg
:target: https://codecov.io/gh/andymccurdy/redis-py
Python 2 Compatibility Note
---------------------------
redis-py 3.5.x will be the last version of redis-py that supports Python 2.
The 3.5.x line will continue to get bug fixes and security patches that
support Python 2 until August 1, 2020. redis-py 4.0 will be the next major
version and will require Python 3.5+.
Installation
------------
redis-py requires a running Redis server. See `Redis's quickstart
<https://redis.io/topics/quickstart>`_ for installation instructions.
redis-py can be installed using `pip` similar to other Python packages. Do not use `sudo`
with `pip`. It is usually good to work in a
`virtualenv <https://virtualenv.pypa.io/en/latest/>`_ or
`venv <https://docs.python.org/3/library/venv.html>`_ to avoid conflicts with other package
managers and Python projects. For a quick introduction see
`Python Virtual Environments in Five Minutes <https://bit.ly/py-env>`_.
To install redis-py, simply:
.. code-block:: bash
$ pip install redis
or from source:
.. code-block:: bash
$ python setup.py install
Getting Started
---------------
.. code-block:: pycon
>>> import redis
>>> r = redis.Redis(host='localhost', port=6379, db=0)
>>> r.set('foo', 'bar')
True
>>> r.get('foo')
b'bar'
By default, all responses are returned as `bytes` in Python 3 and `str` in
Python 2. The user is responsible for decoding to Python 3 strings or Python 2
unicode objects.
If **all** string responses from a client should be decoded, the user can
specify `decode_responses=True` to `Redis.__init__`. In this case, any
Redis command that returns a string type will be decoded with the `encoding`
specified.
Upgrading from redis-py 2.X to 3.0
----------------------------------
redis-py 3.0 introduces many new features but required a number of backwards
incompatible changes to be made in the process. This section attempts to
provide an upgrade path for users migrating from 2.X to 3.0.
Python Version Support
^^^^^^^^^^^^^^^^^^^^^^
redis-py 3.0 supports Python 2.7 and Python 3.5+.
Client Classes: Redis and StrictRedis
^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^^
redis-py 3.0 drops support for the legacy "Redis" client class. "StrictRedis"
has been renamed to "Redis" and an alias named "StrictRedis" is provided so
that users previously using "StrictRedis" can continue to run unchanged.
The 2.X "Redis" class provided alternative implementations of a few commands.
This confused users (rightfully so) and caused a number of support issues. To
make things easier going forward, it was decided to drop support for these
alternate implementations and instead focus on a single client class.
2.X users that are already using StrictRedis don't have to change the class
name. StrictRedis will continue to work for the foreseeable future.
2.X users that are using the Redis class will have to make changes if they
use any of the following commands:
* SETEX: The argument order has changed. The new order is (name, time, value).
* LREM: The argument order has changed. The new order is (name, num, value).
* TTL and PTTL: The return value is now always an int and matches the
official Redis command (>0 indicates the timeout, -1 indicates that the key
exists but that it has no expire time set, -2 indicates that the key does
not exist)
SSL Connections
^^^^^^^^^^^^^^^
redis-py 3.0 changes the default value of the `ssl_cert_reqs` option from
`None` to `'required'`. See
`Issue 1016 <https://github.com/andymccurdy/redis-py/issues/1016>`_. This
change enforces hostname validation when accepting a cert from a remote SSL
terminator. If the terminator doesn't properly set the hostname on the cert
this will cause redis-py 3.0 to raise a ConnectionError.
This check can be disabled by setting `ssl_cert_reqs` to `None`. Note that
doing so removes the security check. Do so at your own risk.
It has been reported that SSL certs received from AWS ElastiCache do not have
proper hostnames and turning off hostname verification is currently required.
MSET, MSETNX and ZADD
^^^^^^^^^^^^^^^^^^^^^
These commands all accept a mapping of key/value pairs. In redis-py 2.X
this mapping could be specified as ``*args`` or as ``**kwargs``. Both of these
styles caused issues when Redis introduced optional flags to ZADD. Relying on
``*args`` caused issues with the optional argument order, especially in Python
2.7. Relying on ``**kwargs`` caused potential collision issues of user keys with
the argument names in the method signature.
To resolve this, redis-py 3.0 has changed these three commands to all accept
a single positional argument named mapping that is expected to be a dict. For
MSET and MSETNX, the dict is a mapping of key-names -> values. For ZADD, the
dict is a mapping of element-names -> score.
MSET, MSETNX and ZADD now look like:
.. code-block:: python
def mset(self, mapping):
def msetnx(self, mapping):
def zadd(self, name, mapping, nx=False, xx=False, ch=False, incr=False):
All 2.X users that use these commands must modify their code to supply
keys and values as a dict to these commands.
ZINCRBY
^^^^^^^
redis-py 2.X accidentally modified the argument order of ZINCRBY, swapping the
order of value and amount. ZINCRBY now looks like:
.. code-block:: python
def zincrby(self, name, amount, value):
All 2.X users that rely on ZINCRBY must swap the order of amount and value
for the command to continue to work as intended.
Encoding of User Input
^^^^^^^^^^^^^^^^^^^^^^
redis-py 3.0 only accepts user data as bytes, strings or numbers (ints, longs
and floats). Attempting to specify a key or a value as any other type will
raise a DataError exception.
redis-py 2.X attempted to coerce any type of input into a string. While
occasionally convenient, this caused all sorts of hidden errors when users
passed boolean values (which were coerced to 'True' or 'False'), a None
value (which was coerced to 'None') or other values, such as user defined
types.
All 2.X users should make sure that the keys and values they pass into
redis-py are either bytes, strings or numbers.
Locks
^^^^^
redis-py 3.0 drops support for the pipeline-based Lock and now only supports
the Lua-based lock. In doing so, LuaLock has been renamed to Lock. This also
means that redis-py Lock objects require Redis server 2.6 or greater.
2.X users that were explicitly referring to "LuaLock" will have to now refer
to "Lock" instead.
Locks as Context Managers
^^^^^^^^^^^^^^^^^^^^^^^^^
redis-py 3.0 now raises a LockError when using a lock as a context manager and
the lock cannot be acquired within the specified timeout. This is more of a
bug fix than a backwards incompatible change. However, given an error is now
raised where none was before, this might alarm some users.
2.X users should make sure they're wrapping their lock code in a try/catch
like this:
.. code-block:: python
try:
with r.lock('my-lock-key', blocking_timeout=5) as lock:
# code you want executed only after the lock has been acquired
except LockError:
# the lock wasn't acquired
API Reference
-------------
The `official Redis command documentation <https://redis.io/commands>`_ does a
great job of explaining each command in detail. redis-py attempts to adhere
to the official command syntax. There are a few exceptions:
* **SELECT**: Not implemented. See the explanation in the Thread Safety section
below.
* **DEL**: 'del' is a reserved keyword in the Python syntax. Therefore redis-py
uses 'delete' instead.
* **MULTI/EXEC**: These are implemented as part of the Pipeline class. The
pipeline is wrapped with the MULTI and EXEC statements by default when it
is executed, which can be disabled by specifying transaction=False.
See more about Pipelines below.
* **SUBSCRIBE/LISTEN**: Similar to pipelines, PubSub is implemented as a separate
class as it places the underlying connection in a state where it can't
execute non-pubsub commands. Calling the pubsub method from the Redis client
will return a PubSub instance where you can subscribe to channels and listen
for messages. You can only call PUBLISH from the Redis client (see
`this comment on issue #151
<https://github.com/andymccurdy/redis-py/issues/151#issuecomment-1545015>`_
for details).
* **SCAN/SSCAN/HSCAN/ZSCAN**: The \*SCAN commands are implemented as they
exist in the Redis documentation. In addition, each command has an equivalent
iterator method. These are purely for convenience so the user doesn't have
to keep track of the cursor while iterating. Use the
scan_iter/sscan_iter/hscan_iter/zscan_iter methods for this behavior.
More Detail
-----------
Connection Pools
^^^^^^^^^^^^^^^^
Behind the scenes, redis-py uses a connection pool to manage connections to
a Redis server. By default, each Redis instance you create will in turn create
its own connection pool. You can override this behavior and use an existing
connection pool by passing an already created connection pool instance to the
connection_pool argument of the Redis class. You may choose to do this in order
to implement client side sharding or have fine-grain control of how
connections are managed.
.. code-block:: pycon
>>> pool = redis.ConnectionPool(host='localhost', port=6379, db=0)
>>> r = redis.Redis(connection_pool=pool)
Connections
^^^^^^^^^^^
ConnectionPools manage a set of Connection instances. redis-py ships with two
types of Connections. The default, Connection, is a normal TCP socket based
connection. The UnixDomainSocketConnection allows for clients running on the
same device as the server to connect via a unix domain socket. To use a
UnixDomainSocketConnection connection, simply pass the unix_socket_path
argument, which is a string to the unix domain socket file. Additionally, make
sure the unixsocket parameter is defined in your redis.conf file. It's
commented out by default.
.. code-block:: pycon
>>> r = redis.Redis(unix_socket_path='/tmp/redis.sock')
You can create your own Connection subclasses as well. This may be useful if
you want to control the socket behavior within an async framework. To
instantiate a client class using your own connection, you need to create
a connection pool, passing your class to the connection_class argument.
Other keyword parameters you pass to the pool will be passed to the class
specified during initialization.
.. code-block:: pycon
>>> pool = redis.ConnectionPool(connection_class=YourConnectionClass,
your_arg='...', ...)
Connections maintain an open socket to the Redis server. Sometimes these
sockets are interrupted or disconnected for a variety of reasons. For example,
network appliances, load balancers and other services that sit between clients
and servers are often configured to kill connections that remain idle for a
given threshold.
When a connection becomes disconnected, the next command issued on that
connection will fail and redis-py will raise a ConnectionError to the caller.
This allows each application that uses redis-py to handle errors in a way
that's fitting for that specific application. However, constant error
handling can be verbose and cumbersome, especially when socket disconnections
happen frequently in many production environments.
To combat this, redis-py can issue regular health checks to assess the
liveliness of a connection just before issuing a command. Users can pass
``health_check_interval=N`` to the Redis or ConnectionPool classes or
as a query argument within a Redis URL. The value of ``health_check_interval``
must be an integer. A value of ``0``, the default, disables health checks.
Any positive integer will enable health checks. Health checks are performed
just before a command is executed if the underlying connection has been idle
for more than ``health_check_interval`` seconds. For example,
``health_check_interval=30`` will ensure that a health check is run on any
connection that has been idle for 30 or more seconds just before a command
is executed on that connection.
If your application is running in an environment that disconnects idle
connections after 30 seconds you should set the ``health_check_interval``
option to a value less than 30.
This option also works on any PubSub connection that is created from a
client with ``health_check_interval`` enabled. PubSub users need to ensure
that ``get_message()`` or ``listen()`` are called more frequently than
``health_check_interval`` seconds. It is assumed that most workloads already
do this.
If your PubSub use case doesn't call ``get_message()`` or ``listen()``
frequently, you should call ``pubsub.check_health()`` explicitly on a
regularly basis.
Parsers
^^^^^^^
Parser classes provide a way to control how responses from the Redis server
are parsed. redis-py ships with two parser classes, the PythonParser and the
HiredisParser. By default, redis-py will attempt to use the HiredisParser if
you have the hiredis module installed and will fallback to the PythonParser
otherwise.
Hiredis is a C library maintained by the core Redis team. Pieter Noordhuis was
kind enough to create Python bindings. Using Hiredis can provide up to a
10x speed improvement in parsing responses from the Redis server. The
performance increase is most noticeable when retrieving many pieces of data,
such as from LRANGE or SMEMBERS operations.
Hiredis is available on PyPI, and can be installed via pip just like redis-py.
.. code-block:: bash
$ pip install hiredis
Response Callbacks
^^^^^^^^^^^^^^^^^^
The client class uses a set of callbacks to cast Redis responses to the
appropriate Python type. There are a number of these callbacks defined on
the Redis client class in a dictionary called RESPONSE_CALLBACKS.
Custom callbacks can be added on a per-instance basis using the
set_response_callback method. This method accepts two arguments: a command
name and the callback. Callbacks added in this manner are only valid on the
instance the callback is added to. If you want to define or override a callback
globally, you should make a subclass of the Redis client and add your callback
to its RESPONSE_CALLBACKS class dictionary.
Response callbacks take at least one parameter: the response from the Redis
server. Keyword arguments may also be accepted in order to further control
how to interpret the response. These keyword arguments are specified during the
command's call to execute_command. The ZRANGE implementation demonstrates the
use of response callback keyword arguments with its "withscores" argument.
Thread Safety
^^^^^^^^^^^^^
Redis client instances can safely be shared between threads. Internally,
connection instances are only retrieved from the connection pool during
command execution, and returned to the pool directly after. Command execution
never modifies state on the client instance.
However, there is one caveat: the Redis SELECT command. The SELECT command
allows you to switch the database currently in use by the connection. That
database remains selected until another is selected or until the connection is
closed. This creates an issue in that connections could be returned to the pool
that are connected to a different database.
As a result, redis-py does not implement the SELECT command on client
instances. If you use multiple Redis databases within the same application, you
should create a separate client instance (and possibly a separate connection
pool) for each database.
It is not safe to pass PubSub or Pipeline objects between threads.
Pipelines
^^^^^^^^^
Pipelines are a subclass of the base Redis class that provide support for
buffering multiple commands to the server in a single request. They can be used
to dramatically increase the performance of groups of commands by reducing the
number of back-and-forth TCP packets between the client and server.
Pipelines are quite simple to use:
.. code-block:: pycon
>>> r = redis.Redis(...)
>>> r.set('bing', 'baz')
>>> # Use the pipeline() method to create a pipeline instance
>>> pipe = r.pipeline()
>>> # The following SET commands are buffered
>>> pipe.set('foo', 'bar')
>>> pipe.get('bing')
>>> # the EXECUTE call sends all buffered commands to the server, returning
>>> # a list of responses, one for each command.
>>> pipe.execute()
[True, b'baz']
For ease of use, all commands being buffered into the pipeline return the
pipeline object itself. Therefore calls can be chained like:
.. code-block:: pycon
>>> pipe.set('foo', 'bar').sadd('faz', 'baz').incr('auto_number').execute()
[True, True, 6]
In addition, pipelines can also ensure the buffered commands are executed
atomically as a group. This happens by default. If you want to disable the
atomic nature of a pipeline but still want to buffer commands, you can turn
off transactions.
.. code-block:: pycon
>>> pipe = r.pipeline(transaction=False)
A common issue occurs when requiring atomic transactions but needing to
retrieve values in Redis prior for use within the transaction. For instance,
let's assume that the INCR command didn't exist and we need to build an atomic
version of INCR in Python.
The completely naive implementation could GET the value, increment it in
Python, and SET the new value back. However, this is not atomic because
multiple clients could be doing this at the same time, each getting the same
value from GET.
Enter the WATCH command. WATCH provides the ability to monitor one or more keys
prior to starting a transaction. If any of those keys change prior the
execution of that transaction, the entire transaction will be canceled and a
WatchError will be raised. To implement our own client-side INCR command, we
could do something like this:
.. code-block:: pycon
>>> with r.pipeline() as pipe:
... while True:
... try:
... # put a WATCH on the key that holds our sequence value
... pipe.watch('OUR-SEQUENCE-KEY')
... # after WATCHing, the pipeline is put into immediate execution
... # mode until we tell it to start buffering commands again.
... # this allows us to get the current value of our sequence
... current_value = pipe.get('OUR-SEQUENCE-KEY')
... next_value = int(current_value) + 1
... # now we can put the pipeline back into buffered mode with MULTI
... pipe.multi()
... pipe.set('OUR-SEQUENCE-KEY', next_value)
... # and finally, execute the pipeline (the set command)
... pipe.execute()
... # if a WatchError wasn't raised during execution, everything
... # we just did happened atomically.
... break
... except WatchError:
... # another client must have changed 'OUR-SEQUENCE-KEY' between
... # the time we started WATCHing it and the pipeline's execution.
... # our best bet is to just retry.
... continue
Note that, because the Pipeline must bind to a single connection for the
duration of a WATCH, care must be taken to ensure that the connection is
returned to the connection pool by calling the reset() method. If the
Pipeline is used as a context manager (as in the example above) reset()
will be called automatically. Of course you can do this the manual way by
explicitly calling reset():
.. code-block:: pycon
>>> pipe = r.pipeline()
>>> while True:
... try:
... pipe.watch('OUR-SEQUENCE-KEY')
... ...
... pipe.execute()
... break
... except WatchError:
... continue
... finally:
... pipe.reset()
A convenience method named "transaction" exists for handling all the
boilerplate of handling and retrying watch errors. It takes a callable that
should expect a single parameter, a pipeline object, and any number of keys to
be WATCHed. Our client-side INCR command above can be written like this,
which is much easier to read:
.. code-block:: pycon
>>> def client_side_incr(pipe):
... current_value = pipe.get('OUR-SEQUENCE-KEY')
... next_value = int(current_value) + 1
... pipe.multi()
... pipe.set('OUR-SEQUENCE-KEY', next_value)
>>>
>>> r.transaction(client_side_incr, 'OUR-SEQUENCE-KEY')
[True]
Be sure to call `pipe.multi()` in the callable passed to `Redis.transaction`
prior to any write commands.
Publish / Subscribe
^^^^^^^^^^^^^^^^^^^
redis-py includes a `PubSub` object that subscribes to channels and listens
for new messages. Creating a `PubSub` object is easy.
.. code-block:: pycon
>>> r = redis.Redis(...)
>>> p = r.pubsub()
Once a `PubSub` instance is created, channels and patterns can be subscribed
to.
.. code-block:: pycon
>>> p.subscribe('my-first-channel', 'my-second-channel', ...)
>>> p.psubscribe('my-*', ...)
The `PubSub` instance is now subscribed to those channels/patterns. The
subscription confirmations can be seen by reading messages from the `PubSub`
instance.
.. code-block:: pycon
>>> p.get_message()
{'pattern': None, 'type': 'subscribe', 'channel': b'my-second-channel', 'data': 1}
>>> p.get_message()
{'pattern': None, 'type': 'subscribe', 'channel': b'my-first-channel', 'data': 2}
>>> p.get_message()
{'pattern': None, 'type': 'psubscribe', 'channel': b'my-*', 'data': 3}
Every message read from a `PubSub` instance will be a dictionary with the
following keys.
* **type**: One of the following: 'subscribe', 'unsubscribe', 'psubscribe',
'punsubscribe', 'message', 'pmessage'
* **channel**: The channel [un]subscribed to or the channel a message was
published to
* **pattern**: The pattern that matched a published message's channel. Will be
`None` in all cases except for 'pmessage' types.
* **data**: The message data. With [un]subscribe messages, this value will be
the number of channels and patterns the connection is currently subscribed
to. With [p]message messages, this value will be the actual published
message.
Let's send a message now.
.. code-block:: pycon
# the publish method returns the number matching channel and pattern
# subscriptions. 'my-first-channel' matches both the 'my-first-channel'
# subscription and the 'my-*' pattern subscription, so this message will
# be delivered to 2 channels/patterns
>>> r.publish('my-first-channel', 'some data')
2
>>> p.get_message()
{'channel': b'my-first-channel', 'data': b'some data', 'pattern': None, 'type': 'message'}
>>> p.get_message()
{'channel': b'my-first-channel', 'data': b'some data', 'pattern': b'my-*', 'type': 'pmessage'}
Unsubscribing works just like subscribing. If no arguments are passed to
[p]unsubscribe, all channels or patterns will be unsubscribed from.
.. code-block:: pycon
>>> p.unsubscribe()
>>> p.punsubscribe('my-*')
>>> p.get_message()
{'channel': b'my-second-channel', 'data': 2, 'pattern': None, 'type': 'unsubscribe'}
>>> p.get_message()
{'channel': b'my-first-channel', 'data': 1, 'pattern': None, 'type': 'unsubscribe'}
>>> p.get_message()
{'channel': b'my-*', 'data': 0, 'pattern': None, 'type': 'punsubscribe'}
redis-py also allows you to register callback functions to handle published
messages. Message handlers take a single argument, the message, which is a
dictionary just like the examples above. To subscribe to a channel or pattern
with a message handler, pass the channel or pattern name as a keyword argument
with its value being the callback function.
When a message is read on a channel or pattern with a message handler, the
message dictionary is created and passed to the message handler. In this case,
a `None` value is returned from get_message() since the message was already
handled.
.. code-block:: pycon
>>> def my_handler(message):
... print('MY HANDLER: ', message['data'])
>>> p.subscribe(**{'my-channel': my_handler})
# read the subscribe confirmation message
>>> p.get_message()
{'pattern': None, 'type': 'subscribe', 'channel': b'my-channel', 'data': 1}
>>> r.publish('my-channel', 'awesome data')
1
# for the message handler to work, we need tell the instance to read data.
# this can be done in several ways (read more below). we'll just use
# the familiar get_message() function for now
>>> message = p.get_message()
MY HANDLER: awesome data
# note here that the my_handler callback printed the string above.
# `message` is None because the message was handled by our handler.
>>> print(message)
None
If your application is not interested in the (sometimes noisy)
subscribe/unsubscribe confirmation messages, you can ignore them by passing
`ignore_subscribe_messages=True` to `r.pubsub()`. This will cause all
subscribe/unsubscribe messages to be read, but they won't bubble up to your
application.
.. code-block:: pycon
>>> p = r.pubsub(ignore_subscribe_messages=True)
>>> p.subscribe('my-channel')
>>> p.get_message() # hides the subscribe message and returns None
>>> r.publish('my-channel', 'my data')
1
>>> p.get_message()
{'channel': b'my-channel', 'data': b'my data', 'pattern': None, 'type': 'message'}
There are three different strategies for reading messages.
The examples above have been using `pubsub.get_message()`. Behind the scenes,
`get_message()` uses the system's 'select' module to quickly poll the
connection's socket. If there's data available to be read, `get_message()` will
read it, format the message and return it or pass it to a message handler. If
there's no data to be read, `get_message()` will immediately return None. This
makes it trivial to integrate into an existing event loop inside your
application.
.. code-block:: pycon
>>> while True:
>>> message = p.get_message()
>>> if message:
>>> # do something with the message
>>> time.sleep(0.001) # be nice to the system :)
Older versions of redis-py only read messages with `pubsub.listen()`. listen()
is a generator that blocks until a message is available. If your application
doesn't need to do anything else but receive and act on messages received from
redis, listen() is an easy way to get up an running.
.. code-block:: pycon
>>> for message in p.listen():
... # do something with the message
The third option runs an event loop in a separate thread.
`pubsub.run_in_thread()` creates a new thread and starts the event loop. The
thread object is returned to the caller of `run_in_thread()`. The caller can
use the `thread.stop()` method to shut down the event loop and thread. Behind
the scenes, this is simply a wrapper around `get_message()` that runs in a
separate thread, essentially creating a tiny non-blocking event loop for you.
`run_in_thread()` takes an optional `sleep_time` argument. If specified, the
event loop will call `time.sleep()` with the value in each iteration of the
loop.
Note: Since we're running in a separate thread, there's no way to handle
messages that aren't automatically handled with registered message handlers.
Therefore, redis-py prevents you from calling `run_in_thread()` if you're
subscribed to patterns or channels that don't have message handlers attached.
.. code-block:: pycon
>>> p.subscribe(**{'my-channel': my_handler})
>>> thread = p.run_in_thread(sleep_time=0.001)
# the event loop is now running in the background processing messages
# when it's time to shut it down...
>>> thread.stop()
A PubSub object adheres to the same encoding semantics as the client instance
it was created from. Any channel or pattern that's unicode will be encoded
using the `charset` specified on the client before being sent to Redis. If the
client's `decode_responses` flag is set the False (the default), the
'channel', 'pattern' and 'data' values in message dictionaries will be byte
strings (str on Python 2, bytes on Python 3). If the client's
`decode_responses` is True, then the 'channel', 'pattern' and 'data' values
will be automatically decoded to unicode strings using the client's `charset`.
PubSub objects remember what channels and patterns they are subscribed to. In
the event of a disconnection such as a network error or timeout, the
PubSub object will re-subscribe to all prior channels and patterns when
reconnecting. Messages that were published while the client was disconnected
cannot be delivered. When you're finished with a PubSub object, call its
`.close()` method to shutdown the connection.
.. code-block:: pycon
>>> p = r.pubsub()
>>> ...
>>> p.close()
The PUBSUB set of subcommands CHANNELS, NUMSUB and NUMPAT are also
supported:
.. code-block:: pycon
>>> r.pubsub_channels()
[b'foo', b'bar']
>>> r.pubsub_numsub('foo', 'bar')
[(b'foo', 9001), (b'bar', 42)]
>>> r.pubsub_numsub('baz')
[(b'baz', 0)]
>>> r.pubsub_numpat()
1204
Monitor
^^^^^^^
redis-py includes a `Monitor` object that streams every command processed
by the Redis server. Use `listen()` on the `Monitor` object to block
until a command is received.
.. code-block:: pycon
>>> r = redis.Redis(...)
>>> with r.monitor() as m:
>>> for command in m.listen():
>>> print(command)
Lua Scripting
^^^^^^^^^^^^^
redis-py supports the EVAL, EVALSHA, and SCRIPT commands. However, there are
a number of edge cases that make these commands tedious to use in real world
scenarios. Therefore, redis-py exposes a Script object that makes scripting
much easier to use.
To create a Script instance, use the `register_script` function on a client
instance passing the Lua code as the first argument. `register_script` returns
a Script instance that you can use throughout your code.
The following trivial Lua script accepts two parameters: the name of a key and
a multiplier value. The script fetches the value stored in the key, multiplies
it with the multiplier value and returns the result.
.. code-block:: pycon
>>> r = redis.Redis()
>>> lua = """
... local value = redis.call('GET', KEYS[1])
... value = tonumber(value)
... return value * ARGV[1]"""
>>> multiply = r.register_script(lua)
`multiply` is now a Script instance that is invoked by calling it like a
function. Script instances accept the following optional arguments:
* **keys**: A list of key names that the script will access. This becomes the
KEYS list in Lua.
* **args**: A list of argument values. This becomes the ARGV list in Lua.
* **client**: A redis-py Client or Pipeline instance that will invoke the
script. If client isn't specified, the client that initially
created the Script instance (the one that `register_script` was
invoked from) will be used.
Continuing the example from above:
.. code-block:: pycon
>>> r.set('foo', 2)
>>> multiply(keys=['foo'], args=[5])
10
The value of key 'foo' is set to 2. When multiply is invoked, the 'foo' key is
passed to the script along with the multiplier value of 5. Lua executes the
script and returns the result, 10.
Script instances can be executed using a different client instance, even one
that points to a completely different Redis server.
.. code-block:: pycon
>>> r2 = redis.Redis('redis2.example.com')
>>> r2.set('foo', 3)
>>> multiply(keys=['foo'], args=[5], client=r2)
15
The Script object ensures that the Lua script is loaded into Redis's script
cache. In the event of a NOSCRIPT error, it will load the script and retry
executing it.
Script objects can also be used in pipelines. The pipeline instance should be
passed as the client argument when calling the script. Care is taken to ensure
that the script is registered in Redis's script cache just prior to pipeline
execution.
.. code-block:: pycon
>>> pipe = r.pipeline()
>>> pipe.set('foo', 5)
>>> multiply(keys=['foo'], args=[5], client=pipe)
>>> pipe.execute()
[True, 25]
Sentinel support
^^^^^^^^^^^^^^^^
redis-py can be used together with `Redis Sentinel <https://redis.io/topics/sentinel>`_
to discover Redis nodes. You need to have at least one Sentinel daemon running
in order to use redis-py's Sentinel support.
Connecting redis-py to the Sentinel instance(s) is easy. You can use a
Sentinel connection to discover the master and slaves network addresses:
.. code-block:: pycon
>>> from redis.sentinel import Sentinel
>>> sentinel = Sentinel([('localhost', 26379)], socket_timeout=0.1)
>>> sentinel.discover_master('mymaster')
('127.0.0.1', 6379)
>>> sentinel.discover_slaves('mymaster')
[('127.0.0.1', 6380)]
You can also create Redis client connections from a Sentinel instance. You can
connect to either the master (for write operations) or a slave (for read-only
operations).
.. code-block:: pycon
>>> master = sentinel.master_for('mymaster', socket_timeout=0.1)
>>> slave = sentinel.slave_for('mymaster', socket_timeout=0.1)
>>> master.set('foo', 'bar')
>>> slave.get('foo')
b'bar'
The master and slave objects are normal Redis instances with their
connection pool bound to the Sentinel instance. When a Sentinel backed client
attempts to establish a connection, it first queries the Sentinel servers to
determine an appropriate host to connect to. If no server is found,
a MasterNotFoundError or SlaveNotFoundError is raised. Both exceptions are
subclasses of ConnectionError.
When trying to connect to a slave client, the Sentinel connection pool will
iterate over the list of slaves until it finds one that can be connected to.
If no slaves can be connected to, a connection will be established with the
master.
See `Guidelines for Redis clients with support for Redis Sentinel
<https://redis.io/topics/sentinel-clients>`_ to learn more about Redis Sentinel.
Scan Iterators
^^^^^^^^^^^^^^
The \*SCAN commands introduced in Redis 2.8 can be cumbersome to use. While
these commands are fully supported, redis-py also exposes the following methods
that return Python iterators for convenience: `scan_iter`, `hscan_iter`,
`sscan_iter` and `zscan_iter`.
.. code-block:: pycon
>>> for key, value in (('A', '1'), ('B', '2'), ('C', '3')):
... r.set(key, value)
>>> for key in r.scan_iter():
... print(key, r.get(key))
A 1
B 2
C 3
Author
^^^^^^
redis-py is developed and maintained by Andy McCurdy (sedrik@gmail.com).
It can be found here: https://github.com/andymccurdy/redis-py
Special thanks to:
* Ludovico Magnocavallo, author of the original Python Redis client, from
which some of the socket code is still used.
* Alexander Solovyov for ideas on the generic response callback system.
* Paul Hubbard for initial packaging support.
venv/lib/python3.7/site-packages/redis-3.5.3.dist-info/RECORD 0 → 100644
View file @ bb846822
redis/__init__.py,sha256=xHeEs2e5uiAwaV0oxJ_DgdOVr8U8Y5OlSCJ9rRbiLkE,1209
redis/_compat.py,sha256=opM78DdCy4D86p9cpN_O81yNgjVDUwOJGLtMS4LL9-0,5698
redis/client.py,sha256=hs1gxHDN9AcpPy1Cpf6yHq4ICtDYofW9XijXpSDeSG0,159611
redis/connection.py,sha256=B5n2unyz5YhSzhsyK9Wa_AXZjT6guxnqHdZcfbe3jqs,55954
redis/exceptions.py,sha256=phjjyJjnebrM82XDzfjtreGnkWIoSNfDZiyoWs3_zQE,1341
redis/lock.py,sha256=VNfWNN46FBwhcPUnFmzC8N8uLuxCsu2YT2drkEzM6_U,11349
redis/sentinel.py,sha256=EhyjT_tZMWKtwuUMMAIRKwfEPL1qBfoldLFQ8tAN1Dg,11710
redis/utils.py,sha256=wG1Ws79_HgIzAALwYwK4CrVLLloVTRPRqjo1gxF4U7U,674
redis-3.5.3.dist-info/LICENSE,sha256=eQFI2MEvijiycHp0viNDMWutEmmV_1SAGhgbiyMboSQ,1074
redis-3.5.3.dist-info/METADATA,sha256=55ufgygbtE8nqMl0UVKD90EZ01zKyemMdxFFOwpubC4,36674
redis-3.5.3.dist-info/WHEEL,sha256=kGT74LWyRUZrL4VgLh6_g12IeVl_9u9ZVhadrgXZUEY,110
redis-3.5.3.dist-info/top_level.txt,sha256=OMAefszlde6ZoOtlM35AWzpRIrwtcqAMHGlRit-w2-4,6
redis-3.5.3.dist-info/RECORD,,
redis-3.5.3.dist-info/INSTALLER,sha256=zuuue4knoyJ-UwPPXg8fezS7VCrXJQrAP7zeNuwvFQg,4
redis/__pycache__/client.cpython-37.pyc,,
redis/__pycache__/exceptions.cpython-37.pyc,,
redis/__pycache__/lock.cpython-37.pyc,,
redis/__pycache__/_compat.cpython-37.pyc,,
redis/__pycache__/connection.cpython-37.pyc,,
redis/__pycache__/sentinel.cpython-37.pyc,,
redis/__pycache__/utils.cpython-37.pyc,,
redis/__pycache__/__init__.cpython-37.pyc,,
venv/lib/python3.7/site-packages/redis-3.5.3.dist-info/WHEEL 0 → 100644
View file @ bb846822
Wheel-Version: 1.0
Generator: bdist_wheel (0.34.2)
Root-Is-Purelib: true
Tag: py2-none-any
Tag: py3-none-any
venv/lib/python3.7/site-packages/redis/__init__.py 0 → 100644
View file @ bb846822
from redis.client import Redis, StrictRedis
from redis.connection import (
BlockingConnectionPool,
ConnectionPool,
Connection,
SSLConnection,
UnixDomainSocketConnection
)
from redis.utils import from_url
from redis.exceptions import (
AuthenticationError,
AuthenticationWrongNumberOfArgsError,
BusyLoadingError,
ChildDeadlockedError,
ConnectionError,
DataError,
InvalidResponse,
PubSubError,
ReadOnlyError,
RedisError,
ResponseError,
TimeoutError,
WatchError
)
def int_or_str(value):
try:
return int(value)
except ValueError:
return value
__version__ = '3.5.3'
VERSION = tuple(map(int_or_str, __version__.split('.')))
__all__ = [
'AuthenticationError',
'AuthenticationWrongNumberOfArgsError',
'BlockingConnectionPool',
'BusyLoadingError',
'ChildDeadlockedError',
'Connection',
'ConnectionError',
'ConnectionPool',
'DataError',
'from_url',
'InvalidResponse',
'PubSubError',
'ReadOnlyError',
'Redis',
'RedisError',
'ResponseError',
'SSLConnection',
'StrictRedis',
'TimeoutError',
'UnixDomainSocketConnection',
'WatchError',
]
venv/lib/python3.7/site-packages/redis/_compat.py 0 → 100644
View file @ bb846822
"""Internal module for Python 2 backwards compatibility."""
# flake8: noqa
import errno
import socket
import sys
def sendall(sock, *args, **kwargs):
return sock.sendall(*args, **kwargs)
def shutdown(sock, *args, **kwargs):
return sock.shutdown(*args, **kwargs)
def ssl_wrap_socket(context, sock, *args, **kwargs):
return context.wrap_socket(sock, *args, **kwargs)
# For Python older than 3.5, retry EINTR.
if sys.version_info[0] < 3 or (sys.version_info[0] == 3 and
sys.version_info[1] < 5):
# Adapted from https://bugs.python.org/review/23863/patch/14532/54418
import time
# Wrapper for handling interruptable system calls.
def _retryable_call(s, func, *args, **kwargs):
# Some modules (SSL) use the _fileobject wrapper directly and
# implement a smaller portion of the socket interface, thus we
# need to let them continue to do so.
timeout, deadline = None, 0.0
attempted = False
try:
timeout = s.gettimeout()
except AttributeError:
pass
if timeout:
deadline = time.time() + timeout
try:
while True:
if attempted and timeout:
now = time.time()
if now >= deadline:
raise socket.error(errno.EWOULDBLOCK, "timed out")
else:
# Overwrite the timeout on the socket object
# to take into account elapsed time.
s.settimeout(deadline - now)
try:
attempted = True
return func(*args, **kwargs)
except socket.error as e:
if e.args[0] == errno.EINTR:
continue
raise
finally:
# Set the existing timeout back for future
# calls.
if timeout:
s.settimeout(timeout)
def recv(sock, *args, **kwargs):
return _retryable_call(sock, sock.recv, *args, **kwargs)
def recv_into(sock, *args, **kwargs):
return _retryable_call(sock, sock.recv_into, *args, **kwargs)
else: # Python 3.5 and above automatically retry EINTR
def recv(sock, *args, **kwargs):
return sock.recv(*args, **kwargs)
def recv_into(sock, *args, **kwargs):
return sock.recv_into(*args, **kwargs)
if sys.version_info[0] < 3:
# In Python 3, the ssl module raises socket.timeout whereas it raises
# SSLError in Python 2. For compatibility between versions, ensure
# socket.timeout is raised for both.
import functools
try:
from ssl import SSLError as _SSLError
except ImportError:
class _SSLError(Exception):
"""A replacement in case ssl.SSLError is not available."""
pass
_EXPECTED_SSL_TIMEOUT_MESSAGES = (
"The handshake operation timed out",
"The read operation timed out",
"The write operation timed out",
)
def _handle_ssl_timeout(func):
@functools.wraps(func)
def wrapper(*args, **kwargs):
try:
return func(*args, **kwargs)
except _SSLError as e:
message = len(e.args) == 1 and unicode(e.args[0]) or ''
if any(x in message for x in _EXPECTED_SSL_TIMEOUT_MESSAGES):
# Raise socket.timeout for compatibility with Python 3.
raise socket.timeout(*e.args)
raise
return wrapper
recv = _handle_ssl_timeout(recv)
recv_into = _handle_ssl_timeout(recv_into)
sendall = _handle_ssl_timeout(sendall)
shutdown = _handle_ssl_timeout(shutdown)
ssl_wrap_socket = _handle_ssl_timeout(ssl_wrap_socket)
if sys.version_info[0] < 3:
from urllib import unquote
from urlparse import parse_qs, urlparse
from itertools import imap, izip
from string import letters as ascii_letters
from Queue import Queue
# special unicode handling for python2 to avoid UnicodeDecodeError
def safe_unicode(obj, *args):
""" return the unicode representation of obj """
try:
return unicode(obj, *args)
except UnicodeDecodeError:
# obj is byte string
ascii_text = str(obj).encode('string_escape')
return unicode(ascii_text)
def iteritems(x):
return x.iteritems()
def iterkeys(x):
return x.iterkeys()
def itervalues(x):
return x.itervalues()
def nativestr(x):
return x if isinstance(x, str) else x.encode('utf-8', 'replace')
def next(x):
return x.next()
unichr = unichr
xrange = xrange
basestring = basestring
unicode = unicode
long = long
BlockingIOError = socket.error
else:
from urllib.parse import parse_qs, unquote, urlparse
from string import ascii_letters
from queue import Queue
def iteritems(x):
return iter(x.items())
def iterkeys(x):
return iter(x.keys())
def itervalues(x):
return iter(x.values())
def nativestr(x):
return x if isinstance(x, str) else x.decode('utf-8', 'replace')
def safe_unicode(value):
if isinstance(value, bytes):
value = value.decode('utf-8', 'replace')
return str(value)
next = next
unichr = chr
imap = map
izip = zip
xrange = range
basestring = str
unicode = str
long = int
BlockingIOError = BlockingIOError
try: # Python 3
from queue import LifoQueue, Empty, Full
except ImportError: # Python 2
from Queue import LifoQueue, Empty, Full
venv/lib/python3.7/site-packages/redis/client.py 0 → 100644
View file @ bb846822
This source diff could not be displayed because it is too large. You can view the blob instead.
venv/lib/python3.7/site-packages/redis/connection.py 0 → 100644
View file @ bb846822
from __future__ import unicode_literals
from distutils.version import StrictVersion
from itertools import chain
from time import time
import errno
import io
import os
import socket
import threading
import warnings
from redis._compat import (xrange, imap, unicode, long,
nativestr, basestring, iteritems,
LifoQueue, Empty, Full, urlparse, parse_qs,
recv, recv_into, unquote, BlockingIOError,
sendall, shutdown, ssl_wrap_socket)
from redis.exceptions import (
AuthenticationError,
AuthenticationWrongNumberOfArgsError,
BusyLoadingError,
ChildDeadlockedError,
ConnectionError,
DataError,
ExecAbortError,
InvalidResponse,
NoPermissionError,
NoScriptError,
ReadOnlyError,
RedisError,
ResponseError,
TimeoutError,
)
from redis.utils import HIREDIS_AVAILABLE
try:
import ssl
ssl_available = True
except ImportError:
ssl_available = False
NONBLOCKING_EXCEPTION_ERROR_NUMBERS = {
BlockingIOError: errno.EWOULDBLOCK,
}
if ssl_available:
if hasattr(ssl, 'SSLWantReadError'):
NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantReadError] = 2
NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLWantWriteError] = 2
else:
NONBLOCKING_EXCEPTION_ERROR_NUMBERS[ssl.SSLError] = 2
# In Python 2.7 a socket.error is raised for a nonblocking read.
# The _compat module aliases BlockingIOError to socket.error to be
# Python 2/3 compatible.
# However this means that all socket.error exceptions need to be handled
# properly within these exception handlers.
# We need to make sure socket.error is included in these handlers and
# provide a dummy error number that will never match a real exception.
if socket.error not in NONBLOCKING_EXCEPTION_ERROR_NUMBERS:
NONBLOCKING_EXCEPTION_ERROR_NUMBERS[socket.error] = -999999
NONBLOCKING_EXCEPTIONS = tuple(NONBLOCKING_EXCEPTION_ERROR_NUMBERS.keys())
if HIREDIS_AVAILABLE:
import hiredis
hiredis_version = StrictVersion(hiredis.__version__)
HIREDIS_SUPPORTS_CALLABLE_ERRORS = \
hiredis_version >= StrictVersion('0.1.3')
HIREDIS_SUPPORTS_BYTE_BUFFER = \
hiredis_version >= StrictVersion('0.1.4')
HIREDIS_SUPPORTS_ENCODING_ERRORS = \
hiredis_version >= StrictVersion('1.0.0')
if not HIREDIS_SUPPORTS_BYTE_BUFFER:
msg = ("redis-py works best with hiredis >= 0.1.4. You're running "
"hiredis %s. Please consider upgrading." % hiredis.__version__)
warnings.warn(msg)
HIREDIS_USE_BYTE_BUFFER = True
# only use byte buffer if hiredis supports it
if not HIREDIS_SUPPORTS_BYTE_BUFFER:
HIREDIS_USE_BYTE_BUFFER = False
SYM_STAR = b'*'
SYM_DOLLAR = b'$'
SYM_CRLF = b'\r\n'
SYM_EMPTY = b''
SERVER_CLOSED_CONNECTION_ERROR = "Connection closed by server."
SENTINEL = object()
class Encoder(object):
"Encode strings to bytes-like and decode bytes-like to strings"
def __init__(self, encoding, encoding_errors, decode_responses):
self.encoding = encoding
self.encoding_errors = encoding_errors
self.decode_responses = decode_responses
def encode(self, value):
"Return a bytestring or bytes-like representation of the value"
if isinstance(value, (bytes, memoryview)):
return value
elif isinstance(value, bool):
# special case bool since it is a subclass of int
raise DataError("Invalid input of type: 'bool'. Convert to a "
"bytes, string, int or float first.")
elif isinstance(value, float):
value = repr(value).encode()
elif isinstance(value, (int, long)):
# python 2 repr() on longs is '123L', so use str() instead
value = str(value).encode()
elif not isinstance(value, basestring):
# a value we don't know how to deal with. throw an error
typename = type(value).__name__
raise DataError("Invalid input of type: '%s'. Convert to a "
"bytes, string, int or float first." % typename)
if isinstance(value, unicode):
value = value.encode(self.encoding, self.encoding_errors)
return value
def decode(self, value, force=False):
"Return a unicode string from the bytes-like representation"
if self.decode_responses or force:
if isinstance(value, memoryview):
value = value.tobytes()
if isinstance(value, bytes):
value = value.decode(self.encoding, self.encoding_errors)
return value
class BaseParser(object):
EXCEPTION_CLASSES = {
'ERR': {
'max number of clients reached': ConnectionError,
'Client sent AUTH, but no password is set': AuthenticationError,
'invalid password': AuthenticationError,
# some Redis server versions report invalid command syntax
# in lowercase
'wrong number of arguments for \'auth\' command':
AuthenticationWrongNumberOfArgsError,
# some Redis server versions report invalid command syntax
# in uppercase
'wrong number of arguments for \'AUTH\' command':
AuthenticationWrongNumberOfArgsError,
},
'EXECABORT': ExecAbortError,
'LOADING': BusyLoadingError,
'NOSCRIPT': NoScriptError,
'READONLY': ReadOnlyError,
'NOAUTH': AuthenticationError,
'NOPERM': NoPermissionError,
}
def parse_error(self, response):
"Parse an error response"
error_code = response.split(' ')[0]
if error_code in self.EXCEPTION_CLASSES:
response = response[len(error_code) + 1:]
exception_class = self.EXCEPTION_CLASSES[error_code]
if isinstance(exception_class, dict):
exception_class = exception_class.get(response, ResponseError)
return exception_class(response)
return ResponseError(response)
class SocketBuffer(object):
def __init__(self, socket, socket_read_size, socket_timeout):
self._sock = socket
self.socket_read_size = socket_read_size
self.socket_timeout = socket_timeout
self._buffer = io.BytesIO()
# number of bytes written to the buffer from the socket
self.bytes_written = 0
# number of bytes read from the buffer
self.bytes_read = 0
@property
def length(self):
return self.bytes_written - self.bytes_read
def _read_from_socket(self, length=None, timeout=SENTINEL,
raise_on_timeout=True):
sock = self._sock
socket_read_size = self.socket_read_size
buf = self._buffer
buf.seek(self.bytes_written)
marker = 0
custom_timeout = timeout is not SENTINEL
try:
if custom_timeout:
sock.settimeout(timeout)
while True:
data = recv(self._sock, socket_read_size)
# an empty string indicates the server shutdown the socket
if isinstance(data, bytes) and len(data) == 0:
raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
buf.write(data)
data_length = len(data)
self.bytes_written += data_length
marker += data_length
if length is not None and length > marker:
continue
return True
except socket.timeout:
if raise_on_timeout:
raise TimeoutError("Timeout reading from socket")
return False
except NONBLOCKING_EXCEPTIONS as ex:
# if we're in nonblocking mode and the recv raises a
# blocking error, simply return False indicating that
# there's no data to be read. otherwise raise the
# original exception.
allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1)
if not raise_on_timeout and ex.errno == allowed:
return False
raise ConnectionError("Error while reading from socket: %s" %
(ex.args,))
finally:
if custom_timeout:
sock.settimeout(self.socket_timeout)
def can_read(self, timeout):
return bool(self.length) or \
self._read_from_socket(timeout=timeout,
raise_on_timeout=False)
def read(self, length):
length = length + 2 # make sure to read the \r\n terminator
# make sure we've read enough data from the socket
if length > self.length:
self._read_from_socket(length - self.length)
self._buffer.seek(self.bytes_read)
data = self._buffer.read(length)
self.bytes_read += len(data)
# purge the buffer when we've consumed it all so it doesn't
# grow forever
if self.bytes_read == self.bytes_written:
self.purge()
return data[:-2]
def readline(self):
buf = self._buffer
buf.seek(self.bytes_read)
data = buf.readline()
while not data.endswith(SYM_CRLF):
# there's more data in the socket that we need
self._read_from_socket()
buf.seek(self.bytes_read)
data = buf.readline()
self.bytes_read += len(data)
# purge the buffer when we've consumed it all so it doesn't
# grow forever
if self.bytes_read == self.bytes_written:
self.purge()
return data[:-2]
def purge(self):
self._buffer.seek(0)
self._buffer.truncate()
self.bytes_written = 0
self.bytes_read = 0
def close(self):
try:
self.purge()
self._buffer.close()
except Exception:
# issue #633 suggests the purge/close somehow raised a
# BadFileDescriptor error. Perhaps the client ran out of
# memory or something else? It's probably OK to ignore
# any error being raised from purge/close since we're
# removing the reference to the instance below.
pass
self._buffer = None
self._sock = None
class PythonParser(BaseParser):
"Plain Python parsing class"
def __init__(self, socket_read_size):
self.socket_read_size = socket_read_size
self.encoder = None
self._sock = None
self._buffer = None
def __del__(self):
try:
self.on_disconnect()
except Exception:
pass
def on_connect(self, connection):
"Called when the socket connects"
self._sock = connection._sock
self._buffer = SocketBuffer(self._sock,
self.socket_read_size,
connection.socket_timeout)
self.encoder = connection.encoder
def on_disconnect(self):
"Called when the socket disconnects"
self._sock = None
if self._buffer is not None:
self._buffer.close()
self._buffer = None
self.encoder = None
def can_read(self, timeout):
return self._buffer and self._buffer.can_read(timeout)
def read_response(self):
raw = self._buffer.readline()
if not raw:
raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
byte, response = raw[:1], raw[1:]
if byte not in (b'-', b'+', b':', b'$', b'*'):
raise InvalidResponse("Protocol Error: %r" % raw)
# server returned an error
if byte == b'-':
response = nativestr(response)
error = self.parse_error(response)
# if the error is a ConnectionError, raise immediately so the user
# is notified
if isinstance(error, ConnectionError):
raise error
# otherwise, we're dealing with a ResponseError that might belong
# inside a pipeline response. the connection's read_response()
# and/or the pipeline's execute() will raise this error if
# necessary, so just return the exception instance here.
return error
# single value
elif byte == b'+':
pass
# int value
elif byte == b':':
response = long(response)
# bulk response
elif byte == b'$':
length = int(response)
if length == -1:
return None
response = self._buffer.read(length)
# multi-bulk response
elif byte == b'*':
length = int(response)
if length == -1:
return None
response = [self.read_response() for i in xrange(length)]
if isinstance(response, bytes):
response = self.encoder.decode(response)
return response
class HiredisParser(BaseParser):
"Parser class for connections using Hiredis"
def __init__(self, socket_read_size):
if not HIREDIS_AVAILABLE:
raise RedisError("Hiredis is not installed")
self.socket_read_size = socket_read_size
if HIREDIS_USE_BYTE_BUFFER:
self._buffer = bytearray(socket_read_size)
def __del__(self):
try:
self.on_disconnect()
except Exception:
pass
def on_connect(self, connection):
self._sock = connection._sock
self._socket_timeout = connection.socket_timeout
kwargs = {
'protocolError': InvalidResponse,
'replyError': self.parse_error,
}
# hiredis < 0.1.3 doesn't support functions that create exceptions
if not HIREDIS_SUPPORTS_CALLABLE_ERRORS:
kwargs['replyError'] = ResponseError
if connection.encoder.decode_responses:
kwargs['encoding'] = connection.encoder.encoding
if HIREDIS_SUPPORTS_ENCODING_ERRORS:
kwargs['errors'] = connection.encoder.encoding_errors
self._reader = hiredis.Reader(**kwargs)
self._next_response = False
def on_disconnect(self):
self._sock = None
self._reader = None
self._next_response = False
def can_read(self, timeout):
if not self._reader:
raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
if self._next_response is False:
self._next_response = self._reader.gets()
if self._next_response is False:
return self.read_from_socket(timeout=timeout,
raise_on_timeout=False)
return True
def read_from_socket(self, timeout=SENTINEL, raise_on_timeout=True):
sock = self._sock
custom_timeout = timeout is not SENTINEL
try:
if custom_timeout:
sock.settimeout(timeout)
if HIREDIS_USE_BYTE_BUFFER:
bufflen = recv_into(self._sock, self._buffer)
if bufflen == 0:
raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
self._reader.feed(self._buffer, 0, bufflen)
else:
buffer = recv(self._sock, self.socket_read_size)
# an empty string indicates the server shutdown the socket
if not isinstance(buffer, bytes) or len(buffer) == 0:
raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
self._reader.feed(buffer)
# data was read from the socket and added to the buffer.
# return True to indicate that data was read.
return True
except socket.timeout:
if raise_on_timeout:
raise TimeoutError("Timeout reading from socket")
return False
except NONBLOCKING_EXCEPTIONS as ex:
# if we're in nonblocking mode and the recv raises a
# blocking error, simply return False indicating that
# there's no data to be read. otherwise raise the
# original exception.
allowed = NONBLOCKING_EXCEPTION_ERROR_NUMBERS.get(ex.__class__, -1)
if not raise_on_timeout and ex.errno == allowed:
return False
raise ConnectionError("Error while reading from socket: %s" %
(ex.args,))
finally:
if custom_timeout:
sock.settimeout(self._socket_timeout)
def read_response(self):
if not self._reader:
raise ConnectionError(SERVER_CLOSED_CONNECTION_ERROR)
# _next_response might be cached from a can_read() call
if self._next_response is not False:
response = self._next_response
self._next_response = False
return response
response = self._reader.gets()
while response is False:
self.read_from_socket()
response = self._reader.gets()
# if an older version of hiredis is installed, we need to attempt
# to convert ResponseErrors to their appropriate types.
if not HIREDIS_SUPPORTS_CALLABLE_ERRORS:
if isinstance(response, ResponseError):
response = self.parse_error(response.args[0])
elif isinstance(response, list) and response and \
isinstance(response[0], ResponseError):
response[0] = self.parse_error(response[0].args[0])
# if the response is a ConnectionError or the response is a list and
# the first item is a ConnectionError, raise it as something bad
# happened
if isinstance(response, ConnectionError):
raise response
elif isinstance(response, list) and response and \
isinstance(response[0], ConnectionError):
raise response[0]
return response
if HIREDIS_AVAILABLE:
DefaultParser = HiredisParser
else:
DefaultParser = PythonParser
class Connection(object):
"Manages TCP communication to and from a Redis server"
def __init__(self, host='localhost', port=6379, db=0, password=None,
socket_timeout=None, socket_connect_timeout=None,
socket_keepalive=False, socket_keepalive_options=None,
socket_type=0, retry_on_timeout=False, encoding='utf-8',
encoding_errors='strict', decode_responses=False,
parser_class=DefaultParser, socket_read_size=65536,
health_check_interval=0, client_name=None, username=None):
self.pid = os.getpid()
self.host = host
self.port = int(port)
self.db = db
self.username = username
self.client_name = client_name
self.password = password
self.socket_timeout = socket_timeout
self.socket_connect_timeout = socket_connect_timeout or socket_timeout
self.socket_keepalive = socket_keepalive
self.socket_keepalive_options = socket_keepalive_options or {}
self.socket_type = socket_type
self.retry_on_timeout = retry_on_timeout
self.health_check_interval = health_check_interval
self.next_health_check = 0
self.encoder = Encoder(encoding, encoding_errors, decode_responses)
self._sock = None
self._parser = parser_class(socket_read_size=socket_read_size)
self._connect_callbacks = []
self._buffer_cutoff = 6000
def __repr__(self):
repr_args = ','.join(['%s=%s' % (k, v) for k, v in self.repr_pieces()])
return '%s<%s>' % (self.__class__.__name__, repr_args)
def repr_pieces(self):
pieces = [
('host', self.host),
('port', self.port),
('db', self.db)
]
if self.client_name:
pieces.append(('client_name', self.client_name))
return pieces
def __del__(self):
try:
self.disconnect()
except Exception:
pass
def register_connect_callback(self, callback):
self._connect_callbacks.append(callback)
def clear_connect_callbacks(self):
self._connect_callbacks = []
def connect(self):
"Connects to the Redis server if not already connected"
if self._sock:
return
try:
sock = self._connect()
except socket.timeout:
raise TimeoutError("Timeout connecting to server")
except socket.error as e:
raise ConnectionError(self._error_message(e))
self._sock = sock
try:
self.on_connect()
except RedisError:
# clean up after any error in on_connect
self.disconnect()
raise
# run any user callbacks. right now the only internal callback
# is for pubsub channel/pattern resubscription
for callback in self._connect_callbacks:
callback(self)
def _connect(self):
"Create a TCP socket connection"
# we want to mimic what socket.create_connection does to support
# ipv4/ipv6, but we want to set options prior to calling
# socket.connect()
err = None
for res in socket.getaddrinfo(self.host, self.port, self.socket_type,
socket.SOCK_STREAM):
family, socktype, proto, canonname, socket_address = res
sock = None
try:
sock = socket.socket(family, socktype, proto)
# TCP_NODELAY
sock.setsockopt(socket.IPPROTO_TCP, socket.TCP_NODELAY, 1)
# TCP_KEEPALIVE
if self.socket_keepalive:
sock.setsockopt(socket.SOL_SOCKET, socket.SO_KEEPALIVE, 1)
for k, v in iteritems(self.socket_keepalive_options):
sock.setsockopt(socket.IPPROTO_TCP, k, v)
# set the socket_connect_timeout before we connect
sock.settimeout(self.socket_connect_timeout)
# connect
sock.connect(socket_address)
# set the socket_timeout now that we're connected
sock.settimeout(self.socket_timeout)
return sock
except socket.error as _:
err = _
if sock is not None:
sock.close()
if err is not None:
raise err
raise socket.error("socket.getaddrinfo returned an empty list")
def _error_message(self, exception):
# args for socket.error can either be (errno, "message")
# or just "message"
if len(exception.args) == 1:
return "Error connecting to %s:%s. %s." % \
(self.host, self.port, exception.args[0])
else:
return "Error %s connecting to %s:%s. %s." % \
(exception.args[0], self.host, self.port, exception.args[1])
def on_connect(self):
"Initialize the connection, authenticate and select a database"
self._parser.on_connect(self)
# if username and/or password are set, authenticate
if self.username or self.password:
if self.username:
auth_args = (self.username, self.password or '')
else:
auth_args = (self.password,)
# avoid checking health here -- PING will fail if we try
# to check the health prior to the AUTH
self.send_command('AUTH', *auth_args, check_health=False)
try:
auth_response = self.read_response()
except AuthenticationWrongNumberOfArgsError:
# a username and password were specified but the Redis
# server seems to be < 6.0.0 which expects a single password
# arg. retry auth with just the password.
# https://github.com/andymccurdy/redis-py/issues/1274
self.send_command('AUTH', self.password, check_health=False)
auth_response = self.read_response()
if nativestr(auth_response) != 'OK':
raise AuthenticationError('Invalid Username or Password')
# if a client_name is given, set it
if self.client_name:
self.send_command('CLIENT', 'SETNAME', self.client_name)
if nativestr(self.read_response()) != 'OK':
raise ConnectionError('Error setting client name')
# if a database is specified, switch to it
if self.db:
self.send_command('SELECT', self.db)
if nativestr(self.read_response()) != 'OK':
raise ConnectionError('Invalid Database')
def disconnect(self):
"Disconnects from the Redis server"
self._parser.on_disconnect()
if self._sock is None:
return
try:
if os.getpid() == self.pid:
shutdown(self._sock, socket.SHUT_RDWR)
self._sock.close()
except socket.error:
pass
self._sock = None
def check_health(self):
"Check the health of the connection with a PING/PONG"
if self.health_check_interval and time() > self.next_health_check:
try:
self.send_command('PING', check_health=False)
if nativestr(self.read_response()) != 'PONG':
raise ConnectionError(
'Bad response from PING health check')
except (ConnectionError, TimeoutError):
self.disconnect()
self.send_command('PING', check_health=False)
if nativestr(self.read_response()) != 'PONG':
raise ConnectionError(
'Bad response from PING health check')
def send_packed_command(self, command, check_health=True):
"Send an already packed command to the Redis server"
if not self._sock:
self.connect()
# guard against health check recursion
if check_health:
self.check_health()
try:
if isinstance(command, str):
command = [command]
for item in command:
sendall(self._sock, item)
except socket.timeout:
self.disconnect()
raise TimeoutError("Timeout writing to socket")
except socket.error as e:
self.disconnect()
if len(e.args) == 1:
errno, errmsg = 'UNKNOWN', e.args[0]
else:
errno = e.args[0]
errmsg = e.args[1]
raise ConnectionError("Error %s while writing to socket. %s." %
(errno, errmsg))
except BaseException:
self.disconnect()
raise
def send_command(self, *args, **kwargs):
"Pack and send a command to the Redis server"
self.send_packed_command(self.pack_command(*args),
check_health=kwargs.get('check_health', True))
def can_read(self, timeout=0):
"Poll the socket to see if there's data that can be read."
sock = self._sock
if not sock:
self.connect()
sock = self._sock
return self._parser.can_read(timeout)
def read_response(self):
"Read the response from a previously sent command"
try:
response = self._parser.read_response()
except socket.timeout:
self.disconnect()
raise TimeoutError("Timeout reading from %s:%s" %
(self.host, self.port))
except socket.error as e:
self.disconnect()
raise ConnectionError("Error while reading from %s:%s : %s" %
(self.host, self.port, e.args))
except BaseException:
self.disconnect()
raise
if self.health_check_interval:
self.next_health_check = time() + self.health_check_interval
if isinstance(response, ResponseError):
raise response
return response
def pack_command(self, *args):
"Pack a series of arguments into the Redis protocol"
output = []
# the client might have included 1 or more literal arguments in
# the command name, e.g., 'CONFIG GET'. The Redis server expects these
# arguments to be sent separately, so split the first argument
# manually. These arguments should be bytestrings so that they are
# not encoded.
if isinstance(args[0], unicode):
args = tuple(args[0].encode().split()) + args[1:]
elif b' ' in args[0]:
args = tuple(args[0].split()) + args[1:]
buff = SYM_EMPTY.join((SYM_STAR, str(len(args)).encode(), SYM_CRLF))
buffer_cutoff = self._buffer_cutoff
for arg in imap(self.encoder.encode, args):
# to avoid large string mallocs, chunk the command into the
# output list if we're sending large values or memoryviews
arg_length = len(arg)
if (len(buff) > buffer_cutoff or arg_length > buffer_cutoff
or isinstance(arg, memoryview)):
buff = SYM_EMPTY.join(
(buff, SYM_DOLLAR, str(arg_length).encode(), SYM_CRLF))
output.append(buff)
output.append(arg)
buff = SYM_CRLF
else:
buff = SYM_EMPTY.join(
(buff, SYM_DOLLAR, str(arg_length).encode(),
SYM_CRLF, arg, SYM_CRLF))
output.append(buff)
return output
def pack_commands(self, commands):
"Pack multiple commands into the Redis protocol"
output = []
pieces = []
buffer_length = 0
buffer_cutoff = self._buffer_cutoff
for cmd in commands:
for chunk in self.pack_command(*cmd):
chunklen = len(chunk)
if (buffer_length > buffer_cutoff or chunklen > buffer_cutoff
or isinstance(chunk, memoryview)):
output.append(SYM_EMPTY.join(pieces))
buffer_length = 0
pieces = []
if chunklen > buffer_cutoff or isinstance(chunk, memoryview):
output.append(chunk)
else:
pieces.append(chunk)
buffer_length += chunklen
if pieces:
output.append(SYM_EMPTY.join(pieces))
return output
class SSLConnection(Connection):
def __init__(self, ssl_keyfile=None, ssl_certfile=None,
ssl_cert_reqs='required', ssl_ca_certs=None,
ssl_check_hostname=False, **kwargs):
if not ssl_available:
raise RedisError("Python wasn't built with SSL support")
super(SSLConnection, self).__init__(**kwargs)
self.keyfile = ssl_keyfile
self.certfile = ssl_certfile
if ssl_cert_reqs is None:
ssl_cert_reqs = ssl.CERT_NONE
elif isinstance(ssl_cert_reqs, basestring):
CERT_REQS = {
'none': ssl.CERT_NONE,
'optional': ssl.CERT_OPTIONAL,
'required': ssl.CERT_REQUIRED
}
if ssl_cert_reqs not in CERT_REQS:
raise RedisError(
"Invalid SSL Certificate Requirements Flag: %s" %
ssl_cert_reqs)
ssl_cert_reqs = CERT_REQS[ssl_cert_reqs]
self.cert_reqs = ssl_cert_reqs
self.ca_certs = ssl_ca_certs
self.check_hostname = ssl_check_hostname
def _connect(self):
"Wrap the socket with SSL support"
sock = super(SSLConnection, self)._connect()
if hasattr(ssl, "create_default_context"):
context = ssl.create_default_context()
context.check_hostname = self.check_hostname
context.verify_mode = self.cert_reqs
if self.certfile and self.keyfile:
context.load_cert_chain(certfile=self.certfile,
keyfile=self.keyfile)
if self.ca_certs:
context.load_verify_locations(self.ca_certs)
sock = ssl_wrap_socket(context, sock, server_hostname=self.host)
else:
# In case this code runs in a version which is older than 2.7.9,
# we want to fall back to old code
sock = ssl_wrap_socket(ssl,
sock,
cert_reqs=self.cert_reqs,
keyfile=self.keyfile,
certfile=self.certfile,
ca_certs=self.ca_certs)
return sock
class UnixDomainSocketConnection(Connection):
def __init__(self, path='', db=0, username=None, password=None,
socket_timeout=None, encoding='utf-8',
encoding_errors='strict', decode_responses=False,
retry_on_timeout=False,
parser_class=DefaultParser, socket_read_size=65536,
health_check_interval=0, client_name=None):
self.pid = os.getpid()
self.path = path
self.db = db
self.username = username
self.client_name = client_name
self.password = password
self.socket_timeout = socket_timeout
self.retry_on_timeout = retry_on_timeout
self.health_check_interval = health_check_interval
self.next_health_check = 0
self.encoder = Encoder(encoding, encoding_errors, decode_responses)
self._sock = None
self._parser = parser_class(socket_read_size=socket_read_size)
self._connect_callbacks = []
self._buffer_cutoff = 6000
def repr_pieces(self):
pieces = [
('path', self.path),
('db', self.db),
]
if self.client_name:
pieces.append(('client_name', self.client_name))
return pieces
def _connect(self):
"Create a Unix domain socket connection"
sock = socket.socket(socket.AF_UNIX, socket.SOCK_STREAM)
sock.settimeout(self.socket_timeout)
sock.connect(self.path)
return sock
def _error_message(self, exception):
# args for socket.error can either be (errno, "message")
# or just "message"
if len(exception.args) == 1:
return "Error connecting to unix socket: %s. %s." % \
(self.path, exception.args[0])
else:
return "Error %s connecting to unix socket: %s. %s." % \
(exception.args[0], self.path, exception.args[1])
FALSE_STRINGS = ('0', 'F', 'FALSE', 'N', 'NO')
def to_bool(value):
if value is None or value == '':
return None
if isinstance(value, basestring) and value.upper() in FALSE_STRINGS:
return False
return bool(value)
URL_QUERY_ARGUMENT_PARSERS = {
'socket_timeout': float,
'socket_connect_timeout': float,
'socket_keepalive': to_bool,
'retry_on_timeout': to_bool,
'max_connections': int,
'health_check_interval': int,
'ssl_check_hostname': to_bool,
}
class ConnectionPool(object):
"Generic connection pool"
@classmethod
def from_url(cls, url, db=None, decode_components=False, **kwargs):
"""
Return a connection pool configured from the given URL.
For example::
redis://[[username]:[password]]@localhost:6379/0
rediss://[[username]:[password]]@localhost:6379/0
unix://[[username]:[password]]@/path/to/socket.sock?db=0
Three URL schemes are supported:
- ```redis://``
<https://www.iana.org/assignments/uri-schemes/prov/redis>`_ creates a
normal TCP socket connection
- ```rediss://``
<https://www.iana.org/assignments/uri-schemes/prov/rediss>`_ creates
a SSL wrapped TCP socket connection
- ``unix://`` creates a Unix Domain Socket connection
There are several ways to specify a database number. The parse function
will return the first specified option:
1. A ``db`` querystring option, e.g. redis://localhost?db=0
2. If using the redis:// scheme, the path argument of the url, e.g.
redis://localhost/0
3. The ``db`` argument to this function.
If none of these options are specified, db=0 is used.
The ``decode_components`` argument allows this function to work with
percent-encoded URLs. If this argument is set to ``True`` all ``%xx``
escapes will be replaced by their single-character equivalents after
the URL has been parsed. This only applies to the ``hostname``,
``path``, ``username`` and ``password`` components.
Any additional querystring arguments and keyword arguments will be
passed along to the ConnectionPool class's initializer. The querystring
arguments ``socket_connect_timeout`` and ``socket_timeout`` if supplied
are parsed as float values. The arguments ``socket_keepalive`` and
``retry_on_timeout`` are parsed to boolean values that accept
True/False, Yes/No values to indicate state. Invalid types cause a
``UserWarning`` to be raised. In the case of conflicting arguments,
querystring arguments always win.
"""
url = urlparse(url)
url_options = {}
for name, value in iteritems(parse_qs(url.query)):
if value and len(value) > 0:
parser = URL_QUERY_ARGUMENT_PARSERS.get(name)
if parser:
try:
url_options[name] = parser(value[0])
except (TypeError, ValueError):
warnings.warn(UserWarning(
"Invalid value for `%s` in connection URL." % name
))
else:
url_options[name] = value[0]
if decode_components:
username = unquote(url.username) if url.username else None
password = unquote(url.password) if url.password else None
path = unquote(url.path) if url.path else None
hostname = unquote(url.hostname) if url.hostname else None
else:
username = url.username or None
password = url.password or None
path = url.path
hostname = url.hostname
# We only support redis://, rediss:// and unix:// schemes.
if url.scheme == 'unix':
url_options.update({
'username': username,
'password': password,
'path': path,
'connection_class': UnixDomainSocketConnection,
})
elif url.scheme in ('redis', 'rediss'):
url_options.update({
'host': hostname,
'port': int(url.port or 6379),
'username': username,
'password': password,
})
# If there's a path argument, use it as the db argument if a
# querystring value wasn't specified
if 'db' not in url_options and path:
try:
url_options['db'] = int(path.replace('/', ''))
except (AttributeError, ValueError):
pass
if url.scheme == 'rediss':
url_options['connection_class'] = SSLConnection
else:
valid_schemes = ', '.join(('redis://', 'rediss://', 'unix://'))
raise ValueError('Redis URL must specify one of the following '
'schemes (%s)' % valid_schemes)
# last shot at the db value
url_options['db'] = int(url_options.get('db', db or 0))
# update the arguments from the URL values
kwargs.update(url_options)
# backwards compatability
if 'charset' in kwargs:
warnings.warn(DeprecationWarning(
'"charset" is deprecated. Use "encoding" instead'))
kwargs['encoding'] = kwargs.pop('charset')
if 'errors' in kwargs:
warnings.warn(DeprecationWarning(
'"errors" is deprecated. Use "encoding_errors" instead'))
kwargs['encoding_errors'] = kwargs.pop('errors')
return cls(**kwargs)
def __init__(self, connection_class=Connection, max_connections=None,
**connection_kwargs):
"""
Create a connection pool. If max_connections is set, then this
object raises redis.ConnectionError when the pool's limit is reached.
By default, TCP connections are created unless connection_class is
specified. Use redis.UnixDomainSocketConnection for unix sockets.
Any additional keyword arguments are passed to the constructor of
connection_class.
"""
max_connections = max_connections or 2 ** 31
if not isinstance(max_connections, (int, long)) or max_connections < 0:
raise ValueError('"max_connections" must be a positive integer')
self.connection_class = connection_class
self.connection_kwargs = connection_kwargs
self.max_connections = max_connections
# a lock to protect the critical section in _checkpid().
# this lock is acquired when the process id changes, such as
# after a fork. during this time, multiple threads in the child
# process could attempt to acquire this lock. the first thread
# to acquire the lock will reset the data structures and lock
# object of this pool. subsequent threads acquiring this lock
# will notice the first thread already did the work and simply
# release the lock.
self._fork_lock = threading.Lock()
self.reset()
def __repr__(self):
return "%s<%s>" % (
type(self).__name__,
repr(self.connection_class(**self.connection_kwargs)),
)
def reset(self):
self._lock = threading.Lock()
self._created_connections = 0
self._available_connections = []
self._in_use_connections = set()
# this must be the last operation in this method. while reset() is
# called when holding _fork_lock, other threads in this process
# can call _checkpid() which compares self.pid and os.getpid() without
# holding any lock (for performance reasons). keeping this assignment
# as the last operation ensures that those other threads will also
# notice a pid difference and block waiting for the first thread to
# release _fork_lock. when each of these threads eventually acquire
# _fork_lock, they will notice that another thread already called
# reset() and they will immediately release _fork_lock and continue on.
self.pid = os.getpid()
def _checkpid(self):
# _checkpid() attempts to keep ConnectionPool fork-safe on modern
# systems. this is called by all ConnectionPool methods that
# manipulate the pool's state such as get_connection() and release().
#
# _checkpid() determines whether the process has forked by comparing
# the current process id to the process id saved on the ConnectionPool
# instance. if these values are the same, _checkpid() simply returns.
#
# when the process ids differ, _checkpid() assumes that the process
# has forked and that we're now running in the child process. the child
# process cannot use the parent's file descriptors (e.g., sockets).
# therefore, when _checkpid() sees the process id change, it calls
# reset() in order to reinitialize the child's ConnectionPool. this
# will cause the child to make all new connection objects.
#
# _checkpid() is protected by self._fork_lock to ensure that multiple
# threads in the child process do not call reset() multiple times.
#
# there is an extremely small chance this could fail in the following
# scenario:
# 1. process A calls _checkpid() for the first time and acquires
# self._fork_lock.
# 2. while holding self._fork_lock, process A forks (the fork()
# could happen in a different thread owned by process A)
# 3. process B (the forked child process) inherits the
# ConnectionPool's state from the parent. that state includes
# a locked _fork_lock. process B will not be notified when
# process A releases the _fork_lock and will thus never be
# able to acquire the _fork_lock.
#
# to mitigate this possible deadlock, _checkpid() will only wait 5
# seconds to acquire _fork_lock. if _fork_lock cannot be acquired in
# that time it is assumed that the child is deadlocked and a
# redis.ChildDeadlockedError error is raised.
if self.pid != os.getpid():
# python 2.7 doesn't support a timeout option to lock.acquire()
# we have to mimic lock timeouts ourselves.
timeout_at = time() + 5
acquired = False
while time() < timeout_at:
acquired = self._fork_lock.acquire(False)
if acquired:
break
if not acquired:
raise ChildDeadlockedError
# reset() the instance for the new process if another thread
# hasn't already done so
try:
if self.pid != os.getpid():
self.reset()
finally:
self._fork_lock.release()
def get_connection(self, command_name, *keys, **options):
"Get a connection from the pool"
self._checkpid()
with self._lock:
try:
connection = self._available_connections.pop()
except IndexError:
connection = self.make_connection()
self._in_use_connections.add(connection)
try:
# ensure this connection is connected to Redis
connection.connect()
# connections that the pool provides should be ready to send
# a command. if not, the connection was either returned to the
# pool before all data has been read or the socket has been
# closed. either way, reconnect and verify everything is good.
try:
if connection.can_read():
raise ConnectionError('Connection has data')
except ConnectionError:
connection.disconnect()
connection.connect()
if connection.can_read():
raise ConnectionError('Connection not ready')
except BaseException:
# release the connection back to the pool so that we don't
# leak it
self.release(connection)
raise
return connection
def get_encoder(self):
"Return an encoder based on encoding settings"
kwargs = self.connection_kwargs
return Encoder(
encoding=kwargs.get('encoding', 'utf-8'),
encoding_errors=kwargs.get('encoding_errors', 'strict'),
decode_responses=kwargs.get('decode_responses', False)
)
def make_connection(self):
"Create a new connection"
if self._created_connections >= self.max_connections:
raise ConnectionError("Too many connections")
self._created_connections += 1
return self.connection_class(**self.connection_kwargs)
def release(self, connection):
"Releases the connection back to the pool"
self._checkpid()
with self._lock:
try:
self._in_use_connections.remove(connection)
except KeyError:
# Gracefully fail when a connection is returned to this pool
# that the pool doesn't actually own
pass
if self.owns_connection(connection):
self._available_connections.append(connection)
else:
# pool doesn't own this connection. do not add it back
# to the pool and decrement the count so that another
# connection can take its place if needed
self._created_connections -= 1
connection.disconnect()
return
def owns_connection(self, connection):
return connection.pid == self.pid
def disconnect(self, inuse_connections=True):
"""
Disconnects connections in the pool
If ``inuse_connections`` is True, disconnect connections that are
current in use, potentially by other threads. Otherwise only disconnect
connections that are idle in the pool.
"""
self._checkpid()
with self._lock:
if inuse_connections:
connections = chain(self._available_connections,
self._in_use_connections)
else:
connections = self._available_connections
for connection in connections:
connection.disconnect()
class BlockingConnectionPool(ConnectionPool):
"""
Thread-safe blocking connection pool::
>>> from redis.client import Redis
>>> client = Redis(connection_pool=BlockingConnectionPool())
It performs the same function as the default
``:py:class: ~redis.connection.ConnectionPool`` implementation, in that,
it maintains a pool of reusable connections that can be shared by
multiple redis clients (safely across threads if required).
The difference is that, in the event that a client tries to get a
connection from the pool when all of connections are in use, rather than
raising a ``:py:class: ~redis.exceptions.ConnectionError`` (as the default
``:py:class: ~redis.connection.ConnectionPool`` implementation does), it
makes the client wait ("blocks") for a specified number of seconds until
a connection becomes available.
Use ``max_connections`` to increase / decrease the pool size::
>>> pool = BlockingConnectionPool(max_connections=10)
Use ``timeout`` to tell it either how many seconds to wait for a connection
to become available, or to block forever:
# Block forever.
>>> pool = BlockingConnectionPool(timeout=None)
# Raise a ``ConnectionError`` after five seconds if a connection is
# not available.
>>> pool = BlockingConnectionPool(timeout=5)
"""
def __init__(self, max_connections=50, timeout=20,
connection_class=Connection, queue_class=LifoQueue,
**connection_kwargs):
self.queue_class = queue_class
self.timeout = timeout
super(BlockingConnectionPool, self).__init__(
connection_class=connection_class,
max_connections=max_connections,
**connection_kwargs)
def reset(self):
# Create and fill up a thread safe queue with ``None`` values.
self.pool = self.queue_class(self.max_connections)
while True:
try:
self.pool.put_nowait(None)
except Full:
break
# Keep a list of actual connection instances so that we can
# disconnect them later.
self._connections = []
# this must be the last operation in this method. while reset() is
# called when holding _fork_lock, other threads in this process
# can call _checkpid() which compares self.pid and os.getpid() without
# holding any lock (for performance reasons). keeping this assignment
# as the last operation ensures that those other threads will also
# notice a pid difference and block waiting for the first thread to
# release _fork_lock. when each of these threads eventually acquire
# _fork_lock, they will notice that another thread already called
# reset() and they will immediately release _fork_lock and continue on.
self.pid = os.getpid()
def make_connection(self):
"Make a fresh connection."
connection = self.connection_class(**self.connection_kwargs)
self._connections.append(connection)
return connection
def get_connection(self, command_name, *keys, **options):
"""
Get a connection, blocking for ``self.timeout`` until a connection
is available from the pool.
If the connection returned is ``None`` then creates a new connection.
Because we use a last-in first-out queue, the existing connections
(having been returned to the pool after the initial ``None`` values
were added) will be returned before ``None`` values. This means we only
create new connections when we need to, i.e.: the actual number of
connections will only increase in response to demand.
"""
# Make sure we haven't changed process.
self._checkpid()
# Try and get a connection from the pool. If one isn't available within
# self.timeout then raise a ``ConnectionError``.
connection = None
try:
connection = self.pool.get(block=True, timeout=self.timeout)
except Empty:
# Note that this is not caught by the redis client and will be
# raised unless handled by application code. If you want never to
raise ConnectionError("No connection available.")
# If the ``connection`` is actually ``None`` then that's a cue to make
# a new connection to add to the pool.
if connection is None:
connection = self.make_connection()
try:
# ensure this connection is connected to Redis
connection.connect()
# connections that the pool provides should be ready to send
# a command. if not, the connection was either returned to the
# pool before all data has been read or the socket has been
# closed. either way, reconnect and verify everything is good.
try:
if connection.can_read():
raise ConnectionError('Connection has data')
except ConnectionError:
connection.disconnect()
connection.connect()
if connection.can_read():
raise ConnectionError('Connection not ready')
except BaseException:
# release the connection back to the pool so that we don't leak it
self.release(connection)
raise
return connection
def release(self, connection):
"Releases the connection back to the pool."
# Make sure we haven't changed process.
self._checkpid()
if not self.owns_connection(connection):
# pool doesn't own this connection. do not add it back
# to the pool. instead add a None value which is a placeholder
# that will cause the pool to recreate the connection if
# its needed.
connection.disconnect()
self.pool.put_nowait(None)
return
# Put the connection back into the pool.
try:
self.pool.put_nowait(connection)
except Full:
# perhaps the pool has been reset() after a fork? regardless,
# we don't want this connection
pass
def disconnect(self):
"Disconnects all connections in the pool."
self._checkpid()
for connection in self._connections:
connection.disconnect()
venv/lib/python3.7/site-packages/redis/exceptions.py 0 → 100644
View file @ bb846822
"Core exceptions raised by the Redis client"
class RedisError(Exception):
pass
class ConnectionError(RedisError):
pass
class TimeoutError(RedisError):
pass
class AuthenticationError(ConnectionError):
pass
class BusyLoadingError(ConnectionError):
pass
class InvalidResponse(RedisError):
pass
class ResponseError(RedisError):
pass
class DataError(RedisError):
pass
class PubSubError(RedisError):
pass
class WatchError(RedisError):
pass
class NoScriptError(ResponseError):
pass
class ExecAbortError(ResponseError):
pass
class ReadOnlyError(ResponseError):
pass
class NoPermissionError(ResponseError):
pass
class LockError(RedisError, ValueError):
"Errors acquiring or releasing a lock"
# NOTE: For backwards compatability, this class derives from ValueError.
# This was originally chosen to behave like threading.Lock.
pass
class LockNotOwnedError(LockError):
"Error trying to extend or release a lock that is (no longer) owned"
pass
class ChildDeadlockedError(Exception):
"Error indicating that a child process is deadlocked after a fork()"
pass
class AuthenticationWrongNumberOfArgsError(ResponseError):
"""
An error to indicate that the wrong number of args
were sent to the AUTH command
"""
pass
venv/lib/python3.7/site-packages/redis/lock.py 0 → 100644
View file @ bb846822
import threading
import time as mod_time
import uuid
from redis.exceptions import LockError, LockNotOwnedError
from redis.utils import dummy
class Lock(object):
"""
A shared, distributed Lock. Using Redis for locking allows the Lock
to be shared across processes and/or machines.
It's left to the user to resolve deadlock issues and make sure
multiple clients play nicely together.
"""
lua_release = None
lua_extend = None
lua_reacquire = None
# KEYS[1] - lock name
# ARGV[1] - token
# return 1 if the lock was released, otherwise 0
LUA_RELEASE_SCRIPT = """
local token = redis.call('get', KEYS[1])
if not token or token ~= ARGV[1] then
return 0
end
redis.call('del', KEYS[1])
return 1
"""
# KEYS[1] - lock name
# ARGV[1] - token
# ARGV[2] - additional milliseconds
# ARGV[3] - "0" if the additional time should be added to the lock's
# existing ttl or "1" if the existing ttl should be replaced
# return 1 if the locks time was extended, otherwise 0
LUA_EXTEND_SCRIPT = """
local token = redis.call('get', KEYS[1])
if not token or token ~= ARGV[1] then
return 0
end
local expiration = redis.call('pttl', KEYS[1])
if not expiration then
expiration = 0
end
if expiration < 0 then
return 0
end
local newttl = ARGV[2]
if ARGV[3] == "0" then
newttl = ARGV[2] + expiration
end
redis.call('pexpire', KEYS[1], newttl)
return 1
"""
# KEYS[1] - lock name
# ARGV[1] - token
# ARGV[2] - milliseconds
# return 1 if the locks time was reacquired, otherwise 0
LUA_REACQUIRE_SCRIPT = """
local token = redis.call('get', KEYS[1])
if not token or token ~= ARGV[1] then
return 0
end
redis.call('pexpire', KEYS[1], ARGV[2])
return 1
"""
def __init__(self, redis, name, timeout=None, sleep=0.1,
blocking=True, blocking_timeout=None, thread_local=True):
"""
Create a new Lock instance named ``name`` using the Redis client
supplied by ``redis``.
``timeout`` indicates a maximum life for the lock.
By default, it will remain locked until release() is called.
``timeout`` can be specified as a float or integer, both representing
the number of seconds to wait.
``sleep`` indicates the amount of time to sleep per loop iteration
when the lock is in blocking mode and another client is currently
holding the lock.
``blocking`` indicates whether calling ``acquire`` should block until
the lock has been acquired or to fail immediately, causing ``acquire``
to return False and the lock not being acquired. Defaults to True.
Note this value can be overridden by passing a ``blocking``
argument to ``acquire``.
``blocking_timeout`` indicates the maximum amount of time in seconds to
spend trying to acquire the lock. A value of ``None`` indicates
continue trying forever. ``blocking_timeout`` can be specified as a
float or integer, both representing the number of seconds to wait.
``thread_local`` indicates whether the lock token is placed in
thread-local storage. By default, the token is placed in thread local
storage so that a thread only sees its token, not a token set by
another thread. Consider the following timeline:
time: 0, thread-1 acquires `my-lock`, with a timeout of 5 seconds.
thread-1 sets the token to "abc"
time: 1, thread-2 blocks trying to acquire `my-lock` using the
Lock instance.
time: 5, thread-1 has not yet completed. redis expires the lock
key.
time: 5, thread-2 acquired `my-lock` now that it's available.
thread-2 sets the token to "xyz"
time: 6, thread-1 finishes its work and calls release(). if the
token is *not* stored in thread local storage, then
thread-1 would see the token value as "xyz" and would be
able to successfully release the thread-2's lock.
In some use cases it's necessary to disable thread local storage. For
example, if you have code where one thread acquires a lock and passes
that lock instance to a worker thread to release later. If thread
local storage isn't disabled in this case, the worker thread won't see
the token set by the thread that acquired the lock. Our assumption
is that these cases aren't common and as such default to using
thread local storage.
"""
self.redis = redis
self.name = name
self.timeout = timeout
self.sleep = sleep
self.blocking = blocking
self.blocking_timeout = blocking_timeout
self.thread_local = bool(thread_local)
self.local = threading.local() if self.thread_local else dummy()
self.local.token = None
self.register_scripts()
def register_scripts(self):
cls = self.__class__
client = self.redis
if cls.lua_release is None:
cls.lua_release = client.register_script(cls.LUA_RELEASE_SCRIPT)
if cls.lua_extend is None:
cls.lua_extend = client.register_script(cls.LUA_EXTEND_SCRIPT)
if cls.lua_reacquire is None:
cls.lua_reacquire = \
client.register_script(cls.LUA_REACQUIRE_SCRIPT)
def __enter__(self):
# force blocking, as otherwise the user would have to check whether
# the lock was actually acquired or not.
if self.acquire(blocking=True):
return self
raise LockError("Unable to acquire lock within the time specified")
def __exit__(self, exc_type, exc_value, traceback):
self.release()
def acquire(self, blocking=None, blocking_timeout=None, token=None):
"""
Use Redis to hold a shared, distributed lock named ``name``.
Returns True once the lock is acquired.
If ``blocking`` is False, always return immediately. If the lock
was acquired, return True, otherwise return False.
``blocking_timeout`` specifies the maximum number of seconds to
wait trying to acquire the lock.
``token`` specifies the token value to be used. If provided, token
must be a bytes object or a string that can be encoded to a bytes
object with the default encoding. If a token isn't specified, a UUID
will be generated.
"""
sleep = self.sleep
if token is None:
token = uuid.uuid1().hex.encode()
else:
encoder = self.redis.connection_pool.get_encoder()
token = encoder.encode(token)
if blocking is None:
blocking = self.blocking
if blocking_timeout is None:
blocking_timeout = self.blocking_timeout
stop_trying_at = None
if blocking_timeout is not None:
stop_trying_at = mod_time.time() + blocking_timeout
while True:
if self.do_acquire(token):
self.local.token = token
return True
if not blocking:
return False
next_try_at = mod_time.time() + sleep
if stop_trying_at is not None and next_try_at > stop_trying_at:
return False
mod_time.sleep(sleep)
def do_acquire(self, token):
if self.timeout:
# convert to milliseconds
timeout = int(self.timeout * 1000)
else:
timeout = None
if self.redis.set(self.name, token, nx=True, px=timeout):
return True
return False
def locked(self):
"""
Returns True if this key is locked by any process, otherwise False.
"""
return self.redis.get(self.name) is not None
def owned(self):
"""
Returns True if this key is locked by this lock, otherwise False.
"""
stored_token = self.redis.get(self.name)
# need to always compare bytes to bytes
# TODO: this can be simplified when the context manager is finished
if stored_token and not isinstance(stored_token, bytes):
encoder = self.redis.connection_pool.get_encoder()
stored_token = encoder.encode(stored_token)
return self.local.token is not None and \
stored_token == self.local.token
def release(self):
"Releases the already acquired lock"
expected_token = self.local.token
if expected_token is None:
raise LockError("Cannot release an unlocked lock")
self.local.token = None
self.do_release(expected_token)
def do_release(self, expected_token):
if not bool(self.lua_release(keys=[self.name],
args=[expected_token],
client=self.redis)):
raise LockNotOwnedError("Cannot release a lock"
" that's no longer owned")
def extend(self, additional_time, replace_ttl=False):
"""
Adds more time to an already acquired lock.
``additional_time`` can be specified as an integer or a float, both
representing the number of seconds to add.
``replace_ttl`` if False (the default), add `additional_time` to
the lock's existing ttl. If True, replace the lock's ttl with
`additional_time`.
"""
if self.local.token is None:
raise LockError("Cannot extend an unlocked lock")
if self.timeout is None:
raise LockError("Cannot extend a lock with no timeout")
return self.do_extend(additional_time, replace_ttl)
def do_extend(self, additional_time, replace_ttl):
additional_time = int(additional_time * 1000)
if not bool(
self.lua_extend(
keys=[self.name],
args=[
self.local.token,
additional_time,
replace_ttl and "1" or "0"
],
client=self.redis,
)
):
raise LockNotOwnedError(
"Cannot extend a lock that's" " no longer owned"
)
return True
def reacquire(self):
"""
Resets a TTL of an already acquired lock back to a timeout value.
"""
if self.local.token is None:
raise LockError("Cannot reacquire an unlocked lock")
if self.timeout is None:
raise LockError("Cannot reacquire a lock with no timeout")
return self.do_reacquire()
def do_reacquire(self):
timeout = int(self.timeout * 1000)
if not bool(self.lua_reacquire(keys=[self.name],
args=[self.local.token, timeout],
client=self.redis)):
raise LockNotOwnedError("Cannot reacquire a lock that's"
" no longer owned")
return True
venv/lib/python3.7/site-packages/redis/sentinel.py 0 → 100644
View file @ bb846822
import random
import weakref
from redis.client import Redis
from redis.connection import ConnectionPool, Connection
from redis.exceptions import (ConnectionError, ResponseError, ReadOnlyError,
TimeoutError)
from redis._compat import iteritems, nativestr, xrange
class MasterNotFoundError(ConnectionError):
pass
class SlaveNotFoundError(ConnectionError):
pass
class SentinelManagedConnection(Connection):
def __init__(self, **kwargs):
self.connection_pool = kwargs.pop('connection_pool')
super(SentinelManagedConnection, self).__init__(**kwargs)
def __repr__(self):
pool = self.connection_pool
s = '%s<service=%s%%s>' % (type(self).__name__, pool.service_name)
if self.host:
host_info = ',host=%s,port=%s' % (self.host, self.port)
s = s % host_info
return s
def connect_to(self, address):
self.host, self.port = address
super(SentinelManagedConnection, self).connect()
if self.connection_pool.check_connection:
self.send_command('PING')
if nativestr(self.read_response()) != 'PONG':
raise ConnectionError('PING failed')
def connect(self):
if self._sock:
return # already connected
if self.connection_pool.is_master:
self.connect_to(self.connection_pool.get_master_address())
else:
for slave in self.connection_pool.rotate_slaves():
try:
return self.connect_to(slave)
except ConnectionError:
continue
raise SlaveNotFoundError # Never be here
def read_response(self):
try:
return super(SentinelManagedConnection, self).read_response()
except ReadOnlyError:
if self.connection_pool.is_master:
# When talking to a master, a ReadOnlyError when likely
# indicates that the previous master that we're still connected
# to has been demoted to a slave and there's a new master.
# calling disconnect will force the connection to re-query
# sentinel during the next connect() attempt.
self.disconnect()
raise ConnectionError('The previous master is now a slave')
raise
class SentinelConnectionPool(ConnectionPool):
"""
Sentinel backed connection pool.
If ``check_connection`` flag is set to True, SentinelManagedConnection
sends a PING command right after establishing the connection.
"""
def __init__(self, service_name, sentinel_manager, **kwargs):
kwargs['connection_class'] = kwargs.get(
'connection_class', SentinelManagedConnection)
self.is_master = kwargs.pop('is_master', True)
self.check_connection = kwargs.pop('check_connection', False)
super(SentinelConnectionPool, self).__init__(**kwargs)
self.connection_kwargs['connection_pool'] = weakref.proxy(self)
self.service_name = service_name
self.sentinel_manager = sentinel_manager
def __repr__(self):
return "%s<service=%s(%s)" % (
type(self).__name__,
self.service_name,
self.is_master and 'master' or 'slave',
)
def reset(self):
super(SentinelConnectionPool, self).reset()
self.master_address = None
self.slave_rr_counter = None
def owns_connection(self, connection):
check = not self.is_master or \
(self.is_master and
self.master_address == (connection.host, connection.port))
parent = super(SentinelConnectionPool, self)
return check and parent.owns_connection(connection)
def get_master_address(self):
master_address = self.sentinel_manager.discover_master(
self.service_name)
if self.is_master:
if self.master_address != master_address:
self.master_address = master_address
# disconnect any idle connections so that they reconnect
# to the new master the next time that they are used.
self.disconnect(inuse_connections=False)
return master_address
def rotate_slaves(self):
"Round-robin slave balancer"
slaves = self.sentinel_manager.discover_slaves(self.service_name)
if slaves:
if self.slave_rr_counter is None:
self.slave_rr_counter = random.randint(0, len(slaves) - 1)
for _ in xrange(len(slaves)):
self.slave_rr_counter = (
self.slave_rr_counter + 1) % len(slaves)
slave = slaves[self.slave_rr_counter]
yield slave
# Fallback to the master connection
try:
yield self.get_master_address()
except MasterNotFoundError:
pass
raise SlaveNotFoundError('No slave found for %r' % (self.service_name))
class Sentinel(object):
"""
Redis Sentinel cluster client
>>> from redis.sentinel import Sentinel
>>> sentinel = Sentinel([('localhost', 26379)], socket_timeout=0.1)
>>> master = sentinel.master_for('mymaster', socket_timeout=0.1)
>>> master.set('foo', 'bar')
>>> slave = sentinel.slave_for('mymaster', socket_timeout=0.1)
>>> slave.get('foo')
b'bar'
``sentinels`` is a list of sentinel nodes. Each node is represented by
a pair (hostname, port).
``min_other_sentinels`` defined a minimum number of peers for a sentinel.
When querying a sentinel, if it doesn't meet this threshold, responses
from that sentinel won't be considered valid.
``sentinel_kwargs`` is a dictionary of connection arguments used when
connecting to sentinel instances. Any argument that can be passed to
a normal Redis connection can be specified here. If ``sentinel_kwargs`` is
not specified, any socket_timeout and socket_keepalive options specified
in ``connection_kwargs`` will be used.
``connection_kwargs`` are keyword arguments that will be used when
establishing a connection to a Redis server.
"""
def __init__(self, sentinels, min_other_sentinels=0, sentinel_kwargs=None,
**connection_kwargs):
# if sentinel_kwargs isn't defined, use the socket_* options from
# connection_kwargs
if sentinel_kwargs is None:
sentinel_kwargs = {
k: v
for k, v in iteritems(connection_kwargs)
if k.startswith('socket_')
}
self.sentinel_kwargs = sentinel_kwargs
self.sentinels = [Redis(hostname, port, **self.sentinel_kwargs)
for hostname, port in sentinels]
self.min_other_sentinels = min_other_sentinels
self.connection_kwargs = connection_kwargs
def __repr__(self):
sentinel_addresses = []
for sentinel in self.sentinels:
sentinel_addresses.append('%s:%s' % (
sentinel.connection_pool.connection_kwargs['host'],
sentinel.connection_pool.connection_kwargs['port'],
))
return '%s<sentinels=[%s]>' % (
type(self).__name__,
','.join(sentinel_addresses))
def check_master_state(self, state, service_name):
if not state['is_master'] or state['is_sdown'] or state['is_odown']:
return False
# Check if our sentinel doesn't see other nodes
if state['num-other-sentinels'] < self.min_other_sentinels:
return False
return True
def discover_master(self, service_name):
"""
Asks sentinel servers for the Redis master's address corresponding
to the service labeled ``service_name``.
Returns a pair (address, port) or raises MasterNotFoundError if no
master is found.
"""
for sentinel_no, sentinel in enumerate(self.sentinels):
try:
masters = sentinel.sentinel_masters()
except (ConnectionError, TimeoutError):
continue
state = masters.get(service_name)
if state and self.check_master_state(state, service_name):
# Put this sentinel at the top of the list
self.sentinels[0], self.sentinels[sentinel_no] = (
sentinel, self.sentinels[0])
return state['ip'], state['port']
raise MasterNotFoundError("No master found for %r" % (service_name,))
def filter_slaves(self, slaves):
"Remove slaves that are in an ODOWN or SDOWN state"
slaves_alive = []
for slave in slaves:
if slave['is_odown'] or slave['is_sdown']:
continue
slaves_alive.append((slave['ip'], slave['port']))
return slaves_alive
def discover_slaves(self, service_name):
"Returns a list of alive slaves for service ``service_name``"
for sentinel in self.sentinels:
try:
slaves = sentinel.sentinel_slaves(service_name)
except (ConnectionError, ResponseError, TimeoutError):
continue
slaves = self.filter_slaves(slaves)
if slaves:
return slaves
return []
def master_for(self, service_name, redis_class=Redis,
connection_pool_class=SentinelConnectionPool, **kwargs):
"""
Returns a redis client instance for the ``service_name`` master.
A SentinelConnectionPool class is used to retrive the master's
address before establishing a new connection.
NOTE: If the master's address has changed, any cached connections to
the old master are closed.
By default clients will be a redis.Redis instance. Specify a
different class to the ``redis_class`` argument if you desire
something different.
The ``connection_pool_class`` specifies the connection pool to use.
The SentinelConnectionPool will be used by default.
All other keyword arguments are merged with any connection_kwargs
passed to this class and passed to the connection pool as keyword
arguments to be used to initialize Redis connections.
"""
kwargs['is_master'] = True
connection_kwargs = dict(self.connection_kwargs)
connection_kwargs.update(kwargs)
return redis_class(connection_pool=connection_pool_class(
service_name, self, **connection_kwargs))
def slave_for(self, service_name, redis_class=Redis,
connection_pool_class=SentinelConnectionPool, **kwargs):
"""
Returns redis client instance for the ``service_name`` slave(s).
A SentinelConnectionPool class is used to retrive the slave's
address before establishing a new connection.
By default clients will be a redis.Redis instance. Specify a
different class to the ``redis_class`` argument if you desire
something different.
The ``connection_pool_class`` specifies the connection pool to use.
The SentinelConnectionPool will be used by default.
All other keyword arguments are merged with any connection_kwargs
passed to this class and passed to the connection pool as keyword
arguments to be used to initialize Redis connections.
"""
kwargs['is_master'] = False
connection_kwargs = dict(self.connection_kwargs)
connection_kwargs.update(kwargs)
return redis_class(connection_pool=connection_pool_class(
service_name, self, **connection_kwargs))
venv/lib/python3.7/site-packages/redis/utils.py 0 → 100644
View file @ bb846822
from contextlib import contextmanager
try:
import hiredis # noqa
HIREDIS_AVAILABLE = True
except ImportError:
HIREDIS_AVAILABLE = False
def from_url(url, db=None, **kwargs):
"""
Returns an active Redis client generated from the given database URL.
Will attempt to extract the database id from the path url fragment, if
none is provided.
"""
from redis.client import Redis
return Redis.from_url(url, db, **kwargs)
@contextmanager
def pipeline(redis_obj):
p = redis_obj.pipeline()
yield p
p.execute()
class dummy(object):
"""
Instances of this class can be used as an attribute container.
"""
pass
Markdown is supported
0% or
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment