Differential D2492 Diff 8898 swh/web/common/utils.py

Changeset View

Standalone View

swh/web/common/utils.py

# Copyright (C) 2017-2019 The Software Heritage developers		# Copyright (C) 2017-2019 The Software Heritage developers
# See the AUTHORS file at the top-level directory of this distribution		# See the AUTHORS file at the top-level directory of this distribution
# License: GNU Affero General Public License version 3, or any later version		# License: GNU Affero General Public License version 3, or any later version
# See top-level LICENSE file for more information		# See top-level LICENSE file for more information

import docutils.parsers.rst		import docutils.parsers.rst
import docutils.utils		import docutils.utils
import re		import re

from datetime import datetime, timezone		from datetime import datetime, timezone
from dateutil import parser as date_parser		from dateutil import parser as date_parser
from dateutil import tz		from dateutil import tz

		from typing import Optional, Dict, Any

from django.urls import reverse as django_reverse		from django.urls import reverse as django_reverse
from django.http import QueryDict		from django.http import QueryDict, HttpRequest

from prometheus_client.registry import CollectorRegistry		from prometheus_client.registry import CollectorRegistry

from rest_framework.authentication import SessionAuthentication		from rest_framework.authentication import SessionAuthentication

from swh.model.exceptions import ValidationError		from swh.model.exceptions import ValidationError
from swh.model.identifiers import (		from swh.model.identifiers import (
persistent_identifier, parse_persistent_identifier,		persistent_identifier, parse_persistent_identifier,
CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT		CONTENT, DIRECTORY, RELEASE, REVISION, SNAPSHOT
)		)

from swh.web.common.exc import BadInputExc		from swh.web.common.exc import BadInputExc
from swh.web.config import get_config		from swh.web.config import get_config


SWH_WEB_METRICS_REGISTRY = CollectorRegistry(auto_describe=True)		SWH_WEB_METRICS_REGISTRY = CollectorRegistry(auto_describe=True)

swh_object_icons = {		swh_object_icons = {
'branch': 'fa fa-code-fork',		'branch': 'fa fa-code-fork',
'branches': 'fa fa-code-fork',		'branches': 'fa fa-code-fork',
'content': 'fa fa-file-text',		'content': 'fa fa-file-text',
'directory': 'fa fa-folder',		'directory': 'fa fa-folder',
'person': 'fa fa-user',		'person': 'fa fa-user',
'revisions history': 'fa fa-history',		'revisions history': 'fa fa-history',
'release': 'fa fa-tag',		'release': 'fa fa-tag',
'releases': 'fa fa-tag',		'releases': 'fa fa-tag',
'revision': 'octicon-git-commit',		'revision': 'octicon-git-commit',
'snapshot': 'fa fa-camera',		'snapshot': 'fa fa-camera',
'visits': 'fa fa-calendar',		'visits': 'fa fa-calendar',
}		}


def reverse(viewname, url_args=None, query_params=None,		def reverse(viewname: str,
current_app=None, urlconf=None):		url_args: Optional[Dict[str, Any]] = None,
		query_params: Optional[Dict[str, Any]] = None,
		current_app: Optional[str] = None,
		urlconf: Optional[str] = None,
		request: Optional[HttpRequest] = None) -> str:
"""An override of django reverse function supporting query parameters.		"""An override of django reverse function supporting query parameters.

Args:		Args:
viewname (str): the name of the django view from which to compute a url		viewname: the name of the django view from which to compute a url
url_args (dict): dictionary of url arguments indexed by their names		url_args: dictionary of url arguments indexed by their names
query_params (dict): dictionary of query parameters to append to the		query_params: dictionary of query parameters to append to the
reversed url		reversed url
current_app (str): the name of the django app tighten to the view		current_app: the name of the django app tighten to the view
urlconf (str): url configuration module		urlconf: url configuration module
		request: build an absolute URI if provided

Returns:		Returns:
str: the url of the requested view with processed arguments and		str: the url of the requested view with processed arguments and
query parameters		query parameters
"""		"""

if url_args:		if url_args:
url_args = {k: v for k, v in url_args.items() if v is not None}		url_args = {k: v for k, v in url_args.items() if v is not None}

url = django_reverse(viewname, urlconf=urlconf, kwargs=url_args,		url = django_reverse(viewname, urlconf=urlconf, kwargs=url_args,
current_app=current_app)		current_app=current_app)

if query_params:		if query_params:
query_params = {k: v for k, v in query_params.items() if v}		query_params = {k: v for k, v in query_params.items() if v}

if query_params and len(query_params) > 0:		if query_params and len(query_params) > 0:
query_dict = QueryDict('', mutable=True)		query_dict = QueryDict('', mutable=True)
for k in sorted(query_params.keys()):		for k in sorted(query_params.keys()):
query_dict[k] = query_params[k]		query_dict[k] = query_params[k]
url += ('?' + query_dict.urlencode(safe='/;:'))		url += ('?' + query_dict.urlencode(safe='/;:'))

		if request is not None:
		url = request.build_absolute_uri(url)

return url		return url


def datetime_to_utc(date):		def datetime_to_utc(date):
"""Returns datetime in UTC without timezone info		"""Returns datetime in UTC without timezone info

Args:		Args:
date (datetime.datetime): input datetime with timezone info		date (datetime.datetime): input datetime with timezone info
▲ Show 20 Lines • Show All 252 Lines • ▼ Show 20 Lines	class EnforceCSRFAuthentication(SessionAuthentication):
"""		"""
Helper class to enforce CSRF validation on a DRF view		Helper class to enforce CSRF validation on a DRF view
when a user is not authenticated.		when a user is not authenticated.
"""		"""
def authenticate(self, request):		def authenticate(self, request):
user = getattr(request._request, 'user', None)		user = getattr(request._request, 'user', None)
self.enforce_csrf(request)		self.enforce_csrf(request)
return (user, None)		return (user, None)


		def resolve_branch_alias(snapshot: Dict[str, Any],
		branch: Optional[Dict[str, Any]]
		) -> Optional[Dict[str, Any]]:
		"""
		Resolve branch alias in snapshot content.

		Args:
		snapshot: a full snapshot content
		branch: a branch alias contained in the snapshot
		Returns:
		The real snapshot branch that got aliased.
		"""
		while branch and branch['target_type'] == 'alias':
		if branch['target'] in snapshot['branches']:
		branch = snapshot['branches'][branch['target']]
		else:
		from swh.web.common import service
		snp = service.lookup_snapshot(
		snapshot['id'], branches_from=branch['target'],
		branches_count=1)
		if snp and branch['target'] in snp['branches']:
		branch = snp['branches'][branch['target']]
		else:
		branch = None
		return branch