FazBrowse GitHub Viewer
|
Trending
|
URL:
|
Home
Tools:
[Download Repo ZIP]
[View Raw Code]
[Original HTTPS Page]
python-gitlab/gitlab/utils.py at master · vishes-shell/python-gitlab · GitHub
vishes-shell
python-gitlab
Repository navigation
Code
Issues
Pull requests
Actions
Projects
Security and quality
Insights
Expand file tree
Breadcrumbs
python-gitlab
/
gitlab
/
utils.py
Copy path
More file actions
More file actions
Latest commit
History
History
History
110 lines (82 loc) · 3.06 KB
Breadcrumbs
python-gitlab
/
gitlab
/
utils.py
Copy path
File metadata and controls
110 lines (82 loc) · 3.06 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
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
109
110
# -*- coding: utf-8 -*-
#
# Copyright (C) 2016-2017 Gauvain Pocentek <gauvain@pocentek.net>
#
# This program is free software: you can redistribute it and/or modify
# it under the terms of the GNU Lesser General Public License as published by
# the Free Software Foundation, either version 3 of the License, or
# (at your option) any later version.
#
# This program is distributed in the hope that it will be useful,
# but WITHOUT ANY WARRANTY; without even the implied warranty of
# MERCHANTABILITY or FITNESS FOR A PARTICULAR PURPOSE. See the
# GNU Lesser General Public License for more details.
#
# You should have received a copy of the GNU Lesser General Public License
# along with this program. If not, see <http://www.gnu.org/licenses/>.
import
asyncio
import
functools
from
inspect
import
getmembers
,
isfunction
from
urllib
.
parse
import
urlparse
class
_StdoutStream
(
object
):
def
__call__
(
self
,
chunk
):
print
(
chunk
)
async
def
aresponse_content
(
response
,
streamed
,
action
):
response
=
await
response
if
streamed
is
False
:
return
response
.
content
if
action
is
None
:
action
=
_StdoutStream
()
async
for
chunk
in
response
.
aiter_bytes
():
if
chunk
:
action
(
chunk
)
def
response_content
(
response
,
streamed
,
action
):
if
asyncio
.
iscoroutine
(
response
):
return
aresponse_content
(
response
,
streamed
,
action
)
if
streamed
is
False
:
return
response
.
content
if
action
is
None
:
action
=
_StdoutStream
()
for
chunk
in
response
.
iter_bytes
():
if
chunk
:
action
(
chunk
)
def
copy_dict
(
dest
,
src
):
for
k
,
v
in
src
.
items
():
if
isinstance
(
v
,
dict
):
# Transform dict values to new attributes. For example:
# custom_attributes: {'foo', 'bar'} =>
# "custom_attributes['foo']": "bar"
for
dict_k
,
dict_v
in
v
.
items
():
dest
[
"%s[%s]"
%
(
k
,
dict_k
)]
=
dict_v
else
:
dest
[
k
]
=
v
def
clean_str_id
(
id
):
return
id
.
replace
(
"/"
,
"%2F"
).
replace
(
"#"
,
"%23"
)
def
sanitized_url
(
url
):
parsed
=
urlparse
(
url
)
new_path
=
parsed
.
path
.
replace
(
"."
,
"%2E"
)
return
parsed
.
_replace
(
path
=
new_path
).
geturl
()
def
remove_none_from_dict
(
data
):
return
{
k
:
v
for
k
,
v
in
data
.
items
()
if
v
is
not
None
}
async
def
async_postprocess
(
self
,
awaitable
,
callback
,
*
args
,
**
kwargs
):
obj
=
await
awaitable
return
callback
(
self
,
obj
,
*
args
,
**
kwargs
)
def
awaitable_postprocess
(
f
):
@
functools
.
wraps
(
f
)
def
wrapped_f
(
self
,
obj
,
*
args
,
**
kwargs
):
if
asyncio
.
iscoroutine
(
obj
):
return
async_postprocess
(
self
,
obj
,
f
,
*
args
,
**
kwargs
)
else
:
return
f
(
self
,
obj
,
*
args
,
**
kwargs
)
return
wrapped_f
def
inherit_docstrings
(
cls
):
"""Inherit docstrings for methods which doesn't have its' own
"""
for
name
,
func
in
getmembers
(
cls
,
isfunction
):
if
func
.
__doc__
:
continue
for
parent
in
cls
.
__mro__
[
1
:]:
if
hasattr
(
parent
,
name
):
func
.
__doc__
=
getattr
(
parent
,
name
).
__doc__
break
return
cls
Back
|
FazBrowse Home
|
New Git URL