forked from readthedocs/readthedocs.org
-
Notifications
You must be signed in to change notification settings - Fork 0
/
utils.py
101 lines (83 loc) · 3 KB
/
utils.py
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
"""Utilities for the builds app."""
from time import monotonic
from contextlib import contextmanager
from django.core.cache import cache
from readthedocs.builds.constants import EXTERNAL
from readthedocs.projects.constants import (
BITBUCKET_REGEXS,
GITHUB_PULL_REQUEST_URL,
GITHUB_REGEXS,
GITLAB_MERGE_REQUEST_URL,
GITLAB_REGEXS,
)
LOCK_EXPIRE = 60 * 180 # Lock expires in 3 hours
def get_github_username_repo(url):
if 'github' in url:
for regex in GITHUB_REGEXS:
match = regex.search(url)
if match:
return match.groups()
return (None, None)
def get_bitbucket_username_repo(url=None):
if 'bitbucket' in url:
for regex in BITBUCKET_REGEXS:
match = regex.search(url)
if match:
return match.groups()
return (None, None)
def get_gitlab_username_repo(url=None):
if 'gitlab' in url:
for regex in GITLAB_REGEXS:
match = regex.search(url)
if match:
return match.groups()
return (None, None)
def get_vcs_url(*, project, version_type, version_name):
"""
Generate VCS (github, gitlab, bitbucket) URL for this version.
Example: https://github.com/rtfd/readthedocs.org/tree/3.4.2/.
External version example: https://github.com/rtfd/readthedocs.org/pull/99/.
"""
if version_type == EXTERNAL:
if 'github' in project.repo:
user, repo = get_github_username_repo(project.repo)
return GITHUB_PULL_REQUEST_URL.format(
user=user,
repo=repo,
number=version_name,
)
if 'gitlab' in project.repo:
user, repo = get_gitlab_username_repo(project.repo)
return GITLAB_MERGE_REQUEST_URL.format(
user=user,
repo=repo,
number=version_name,
)
# TODO: Add VCS URL for BitBucket.
return ''
url = ''
if ('github' in project.repo) or ('gitlab' in project.repo):
url = f'/tree/{version_name}/'
elif 'bitbucket' in project.repo:
url = f'/src/{version_name}'
# TODO: improve this replacing
return project.repo.replace('git://', 'https://').replace('.git', '') + url
@contextmanager
def memcache_lock(lock_id, oid):
"""
Create a lock using django's cache for running a celery task.
From http://docs.celeryproject.org/en/latest/tutorials/task-cookbook.html#cookbook-task-serial
"""
timeout_at = monotonic() + LOCK_EXPIRE - 3
# cache.add fails if the key already exists
status = cache.add(lock_id, oid, LOCK_EXPIRE)
try:
yield status
finally:
# memcache delete is very slow, but we have to use it to take
# advantage of using add() for atomic locking
if monotonic() < timeout_at:
# don't release the lock if we exceeded the timeout
# to lessen the chance of releasing an expired lock
# owned by someone else.
cache.delete(lock_id)