FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
stream-python/stream/utils.py at main · GetStream/stream-python · GitHub
GetStream
stream-python
Repository navigation
Code
Issues
2
(2)
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
stream-python
/
stream
/
utils.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
79 lines (62 loc) · 2.25 KB
Breadcrumbs
stream-python
/
stream
/
utils.py
Copy path
File metadata and controls
79 lines (62 loc) · 2.25 KB
Raw
Copy raw file
Download raw file
Open symbols panel
Edit and raw actions
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
import
re
valid_re
=
re
.
compile
(
r"^[\w-]+$"
)
def
validate_feed_id
(
feed_id
):
"""
Validates the input is in the format of user:1
:param feed_id: a feed such as user:1
Raises ValueError if the format doesn't match
"""
feed_id
=
str
(
feed_id
)
if
len
(
feed_id
.
split
(
":"
))
!=
2
:
msg
=
(
f"Invalid feed_id spec
{
feed_id
}
, "
f"please specify the feed_id as feed_slug:feed_id"
)
raise
ValueError
(
msg
)
feed_slug
,
user_id
=
feed_id
.
split
(
":"
)
validate_feed_slug
(
feed_slug
)
validate_user_id
(
user_id
)
return
feed_id
def
validate_feed_slug
(
feed_slug
):
"""
Validates the feed slug
"""
feed_slug
=
str
(
feed_slug
)
if
not
valid_re
.
match
(
feed_slug
):
msg
=
f"Invalid feed slug
{
feed_slug
}
, please only use letters, numbers and _"
raise
ValueError
(
msg
)
return
feed_slug
def
validate_user_id
(
user_id
):
"""
Validates the user id
"""
user_id
=
str
(
user_id
)
if
not
valid_re
.
match
(
user_id
):
msg
=
f"Invalid user id
{
user_id
}
, please only use letters, numbers and _"
raise
ValueError
(
msg
)
return
user_id
def
validate_foreign_id_time
(
foreign_id_time
):
if
not
isinstance
(
foreign_id_time
, (
list
,
tuple
)):
raise
ValueError
(
"foreign_id_time should be a list of tuples"
)
for
v
in
foreign_id_time
:
if
not
isinstance
(
v
, (
list
,
tuple
)):
raise
ValueError
(
"foreign_id_time elements should be lists or tuples"
)
if
len
(
v
)
!=
2
:
raise
ValueError
(
"foreign_id_time elements should have two elements"
)
def
get_reaction_params
(
reactions
):
if
reactions
is
not
None
and
not
isinstance
(
reactions
, (
dict
,)):
raise
TypeError
(
"reactions argument should be a dictionary"
)
params
=
{}
if
reactions
is
not
None
:
if
reactions
.
get
(
"own"
):
params
[
"withOwnReactions"
]
=
True
if
reactions
.
get
(
"recent"
):
params
[
"withRecentReactions"
]
=
True
if
reactions
.
get
(
"counts"
):
params
[
"withReactionCounts"
]
=
True
kinds
=
reactions
.
get
(
"kinds"
)
if
kinds
:
if
isinstance
(
kinds
,
list
):
kinds
=
","
.
join
(
k
.
strip
()
for
k
in
kinds
if
k
.
strip
())
params
[
"reactionKindsFilter"
]
=
kinds
return
params
Back
|
FazBrowse Home
|
New Git URL