-
Notifications
You must be signed in to change notification settings - Fork 37
Expand file tree
/
Copy pathpush.py
More file actions
108 lines (88 loc) · 3.52 KB
/
Copy pathpush.py
File metadata and controls
108 lines (88 loc) · 3.52 KB
1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72
73
74
75
76
77
78
79
80
81
82
83
84
85
86
87
88
89
90
91
92
93
94
95
96
97
98
99
100
101
102
103
104
105
106
107
108
# coding: utf-8
from __future__ import absolute_import
from __future__ import division
from __future__ import print_function
from __future__ import unicode_literals
import arrow
import dateutil.tz as tz
from leancloud.object_ import Object
from leancloud.errors import LeanCloudError
from leancloud import client
__author__ = "asaka <lan@leancloud.rocks>"
class Installation(Object):
pass
class Notification(Object):
def fetch(self, *args, **kwargs):
"""同步服务器的 Notification 数据
"""
response = client.get("/tables/Notifications/{0}".format(self.id))
self._update_data(response.json())
def save(self, *args, **kwargs):
raise LeanCloudError(code=1, error="Notification does not support modify")
def _encode_time(time):
tzinfo = time.tzinfo
if tzinfo is None:
tzinfo = tz.tzlocal()
return arrow.get(time, tzinfo).to("utc").format("YYYY-MM-DDTHH:mm:ss.SSS") + "Z"
def send(
data,
channels=None,
push_time=None,
expiration_time=None,
expiration_interval=None,
where=None,
cql=None,
flow_control=None,
prod=None,
):
"""
发送推送消息。返回结果为此条推送对应的 _Notification 表中的对象，但是如果需要使用其中的数据，需要调用 fetch() 方法将数据同步至本地。
:param channels: 需要推送的频道
:type channels: list or tuple
:param push_time: 推送的时间
:type push_time: datetime
:param expiration_time: 消息过期的绝对日期时间
:type expiration_time: datetime
:param expiration_interval: 消息过期的相对时间，从调用 API 的时间开始算起，单位是秒
:type expiration_interval: int
:param where: 一个查询 _Installation 表的查询条件 leancloud.Query 对象
:type where: leancloud.Query
:param cql: 一个查询 _Installation 表的查询条件 CQL 语句
:type cql: string_types
:param data: 推送给设备的具体信息，详情查看 https://leancloud.cn/docs/push_guide.html#消息内容_Data
:rtype: Notification
:param flow_control: 不为 None 时开启平滑推送，值为每秒推送的目标终端用户数。开启时指定低于 1000 的值，按 1000 计。
:type: flow_control: int
:param prod: 仅对 iOS 推送有效，设置将推送发至 APNs 的开发环境（dev）还是生产环境（prod）。
:type: prod: string
"""
if expiration_interval and expiration_time:
raise TypeError("Both expiration_time and expiration_interval can't be set")
params = {
"data": data,
}
if prod is None:
if client.USE_PRODUCTION == "0":
params["prod"] = "dev"
else:
params["prod"] = prod
if channels:
params["channels"] = channels
if push_time:
params["push_time"] = _encode_time(push_time)
if expiration_time:
params["expiration_time"] = _encode_time(expiration_time)
if expiration_interval:
params["expiration_interval"] = expiration_interval
if where:
params["where"] = where.dump().get("where", {})
if cql:
params["cql"] = cql
# Do not change this to `if flow_control`, because 0 is falsy in Python,
# but `flow_control = 0` will enable smooth push,
# and it is in fact equivalent to `flow_control = 1000`.
if flow_control is not None:
params["flow_control"] = flow_control
result = client.post("/push", params=params).json()
notification = Notification.create_without_data(result["objectId"])
return notification