diff --git a/mypy.ini b/mypy.ini index a48107ac2..278e3a956 100644 --- a/mypy.ini +++ b/mypy.ini @@ -1,6 +1,19 @@ [mypy] -# TODO: enable more flags like in https://github.com/ocf/slackbridge/blob/master/mypy.ini show_traceback = True ignore_missing_imports = True +check_untyped_defs = True + plugins = mypy_django_plugin.main + +disallow_untyped_defs = True +disallow_untyped_calls = True +disallow_any_generics = True + +warn_no_return = True +warn_redundant_casts = True +warn_unused_configs = True +warn_unused_ignores = True + +[mypy.plugins.django-stubs] +django_settings_module = ocfweb.settings diff --git a/ocfweb/about/lab.py b/ocfweb/about/lab.py index 7385dda15..a48ccfef1 100644 --- a/ocfweb/about/lab.py +++ b/ocfweb/about/lab.py @@ -1,7 +1,9 @@ +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render -def lab_open_source(request): +def lab_open_source(request: HttpRequest) -> HttpResponse: return render( request, 'about/lab-open-source.html', @@ -11,7 +13,7 @@ def lab_open_source(request): ) -def lab_vote(request): +def lab_vote(request: HttpRequest) -> HttpResponse: return render( request, 'about/lab-vote.html', diff --git a/ocfweb/about/staff.py b/ocfweb/about/staff.py index 1ebc634f5..d9c7a3e06 100644 --- a/ocfweb/about/staff.py +++ b/ocfweb/about/staff.py @@ -1,7 +1,9 @@ +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render -def about_staff(request): +def about_staff(request: HttpRequest) -> HttpResponse: return render( request, 'about/staff.html', diff --git a/ocfweb/account/chpass.py b/ocfweb/account/chpass.py index 09633ab46..86f9e4981 100644 --- a/ocfweb/account/chpass.py +++ b/ocfweb/account/chpass.py @@ -1,4 +1,10 @@ +from typing import Any +from typing import Iterator +from typing import List + from django import forms +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from django.template.loader import render_to_string from django.utils.safestring import mark_safe @@ -15,15 +21,14 @@ from ocfweb.component.celery import change_password as change_password_task from ocfweb.component.forms import Form - CALLINK_ERROR_MSG = ( "Couldn't connect to CalLink API. Resetting group " 'account passwords online is unavailable.' ) -def get_accounts_signatory_for(calnet_uid): - def flatten(lst): +def get_accounts_signatory_for(calnet_uid: str) -> List[Any]: + def flatten(lst: Iterator[Any]) -> List[Any]: return [item for sublist in lst for item in sublist] group_accounts = flatten( @@ -40,7 +45,7 @@ def flatten(lst): return group_accounts -def get_accounts_for(calnet_uid): +def get_accounts_for(calnet_uid: str) -> List[Any]: accounts = users_by_calnet_uid(calnet_uid) if calnet_uid in TESTER_CALNET_UIDS: @@ -51,7 +56,7 @@ def get_accounts_for(calnet_uid): @calnet_required -def change_password(request): +def change_password(request: HttpRequest) -> HttpResponse: calnet_uid = request.session['calnet_uid'] error = None accounts = get_accounts_for(calnet_uid) @@ -117,19 +122,20 @@ def change_password(request): class ChpassForm(Form): - - def __init__(self, ocf_accounts, calnet_uid, *args, **kwargs): + # fix self.fields.keyOrder type error in mypy + field_order = [ + 'ocf_account', + 'new_password', + 'confirm_password', + ] + + def __init__(self, ocf_accounts: List[str], calnet_uid: str, *args: Any, **kwargs: Any) -> None: super().__init__(*args, **kwargs) self.calnet_uid = calnet_uid self.fields['ocf_account'] = forms.ChoiceField( choices=[(x, x) for x in ocf_accounts], label='OCF account', ) - self.fields.keyOrder = [ - 'ocf_account', - 'new_password', - 'confirm_password', - ] new_password = forms.CharField( widget=forms.PasswordInput, @@ -141,7 +147,7 @@ def __init__(self, ocf_accounts, calnet_uid, *args, **kwargs): label='Confirm password', ) - def clean_ocf_account(self): + def clean_ocf_account(self) -> str: data = self.cleaned_data['ocf_account'] if not user_exists(data): raise forms.ValidationError('OCF user account does not exist.') @@ -161,7 +167,7 @@ def clean_ocf_account(self): return data - def clean_confirm_password(self): + def clean_confirm_password(self) -> str: new_password = self.cleaned_data.get('new_password') confirm_password = self.cleaned_data.get('confirm_password') diff --git a/ocfweb/account/commands.py b/ocfweb/account/commands.py index 056e2e18e..c71aa69bf 100644 --- a/ocfweb/account/commands.py +++ b/ocfweb/account/commands.py @@ -1,5 +1,7 @@ from django import forms from django.forms import widgets +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from paramiko import AuthenticationException from paramiko import SSHClient @@ -8,7 +10,7 @@ from ocfweb.component.forms import Form -def commands(request): +def commands(request: HttpRequest) -> HttpResponse: command_to_run = '' output = '' error = '' diff --git a/ocfweb/account/recommender.py b/ocfweb/account/recommender.py index 170e7e7cd..d57a08d77 100644 --- a/ocfweb/account/recommender.py +++ b/ocfweb/account/recommender.py @@ -1,15 +1,17 @@ from random import randint +from typing import Any +from typing import List from ocflib.account.creation import validate_username from ocflib.account.creation import ValidationError from ocflib.account.creation import ValidationWarning -def recommend(real_name, n): - name_fields = [name.lower() for name in real_name.split()] +def recommend(real_name: str, n: int) -> List[Any]: + name_fields: List[str] = [name.lower() for name in real_name.split()] # Can reimplement name_field_abbrevs to only remove vowels or consonants - name_field_abbrevs = [[] for i in range(len(name_fields))] + name_field_abbrevs: List[List[str]] = [[] for i in range(len(name_fields))] for i in range(len(name_fields)): name_field = name_fields[i] for j in range(1, len(name_field) + 1): @@ -23,7 +25,7 @@ def recommend(real_name, n): new_unvalidated_recs.append(rec + name_field_abbrev) unvalidated_recs = new_unvalidated_recs - validated_recs = [] + validated_recs: List[Any] = [] while len(validated_recs) < n and len(unvalidated_recs) > 0: rec = unvalidated_recs.pop(randint(0, len(unvalidated_recs) - 1)) try: diff --git a/ocfweb/account/register.py b/ocfweb/account/register.py index 978d1d848..187728a9f 100644 --- a/ocfweb/account/register.py +++ b/ocfweb/account/register.py @@ -1,10 +1,13 @@ +from typing import Union + import ocflib.account.search as search import ocflib.account.validators as validators import ocflib.misc.validators import ocflib.ucb.directory as directory from Crypto.PublicKey import RSA from django import forms -from django.core.exceptions import NON_FIELD_ERRORS +from django.http import HttpRequest +from django.http import HttpResponse from django.http import HttpResponseBadRequest from django.http import HttpResponseRedirect from django.http import JsonResponse @@ -29,7 +32,7 @@ @calnet_required -def request_account(request): +def request_account(request: HttpRequest) -> Union[HttpResponseRedirect, HttpResponse]: calnet_uid = request.session['calnet_uid'] status = 'new_request' @@ -87,10 +90,10 @@ def request_account(request): if isinstance(task.result, NewAccountResponse): if task.result.status == NewAccountResponse.REJECTED: status = 'has_errors' - form._errors[NON_FIELD_ERRORS] = form.error_class(task.result.errors) + form.add_error('NON_FIELD_ERRORS', task.result.errors) elif task.result.status == NewAccountResponse.FLAGGED: status = 'has_warnings' - form._errors[NON_FIELD_ERRORS] = form.error_class(task.result.errors) + form.add_error('NON_FIELD_ERRORS', task.result.errors) elif task.result.status == NewAccountResponse.PENDING: return HttpResponseRedirect(reverse('account_pending')) else: @@ -114,7 +117,7 @@ def request_account(request): ) -def recommend(request): +def recommend(request: HttpRequest) -> Union[JsonResponse, HttpResponseBadRequest]: real_name = request.GET.get('real_name', None) if real_name is None: return HttpResponseBadRequest('No real_name in recommend request') @@ -127,7 +130,7 @@ def recommend(request): ) -def validate(request): +def validate(request: HttpRequest) -> Union[HttpResponseBadRequest, JsonResponse]: real_name = request.GET.get('real_name', None) if real_name is None: return HttpResponseBadRequest('No real_name in validate request') @@ -149,7 +152,7 @@ def validate(request): }) -def wait_for_account(request): +def wait_for_account(request: HttpRequest) -> Union[HttpResponse, HttpResponseRedirect]: if 'approve_task_id' not in request.session: return render( request, @@ -180,11 +183,11 @@ def wait_for_account(request): return render(request, 'account/register/wait/error-probably-not-created.html', {}) -def account_pending(request): +def account_pending(request: HttpRequest) -> HttpResponse: return render(request, 'account/register/pending.html', {'title': 'Account request pending'}) -def account_created(request): +def account_created(request: HttpRequest) -> HttpResponse: return render(request, 'account/register/success.html', {'title': 'Account request successful'}) @@ -232,7 +235,7 @@ class ApproveForm(Form): }, ) - def clean_verify_password(self): + def clean_verify_password(self) -> str: password = self.cleaned_data.get('password') verify_password = self.cleaned_data.get('verify_password') @@ -241,7 +244,7 @@ def clean_verify_password(self): raise forms.ValidationError("Your passwords don't match.") return verify_password - def clean_verify_contact_email(self): + def clean_verify_contact_email(self) -> str: email = self.cleaned_data.get('contact_email') verify_contact_email = self.cleaned_data.get('verify_contact_email') @@ -250,7 +253,8 @@ def clean_verify_contact_email(self): raise forms.ValidationError("Your emails don't match.") return verify_contact_email - def clean(self): + # clean incompatible with supertype BaseForm which is defined in django. + def clean(self) -> None: # type: ignore cleaned_data = super().clean() # validate password (requires username to check similarity) diff --git a/ocfweb/account/templatetags/vhost_mail.py b/ocfweb/account/templatetags/vhost_mail.py index 4254f1ac1..c0c5d5544 100644 --- a/ocfweb/account/templatetags/vhost_mail.py +++ b/ocfweb/account/templatetags/vhost_mail.py @@ -1,8 +1,10 @@ +from typing import List + from django import template register = template.Library() @register.filter -def address_to_parts(address): +def address_to_parts(address: str) -> List[str]: return address.split('@') diff --git a/ocfweb/account/vhost.py b/ocfweb/account/vhost.py index 6b6d1c130..052e2e7b5 100644 --- a/ocfweb/account/vhost.py +++ b/ocfweb/account/vhost.py @@ -2,9 +2,12 @@ import re import socket from textwrap import dedent +from typing import Any from django import forms from django.conf import settings +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import redirect from django.shortcuts import render from django.urls import reverse @@ -23,18 +26,18 @@ from ocfweb.component.session import logged_in_user -def available_domain(domain): +def available_domain(domain: str) -> bool: if not re.match(r'^[a-zA-Z0-9]+\.berkeley\.edu$', domain): return False return not host_exists(domain) -def valid_domain_external(domain): +def valid_domain_external(domain: str) -> bool: return bool(re.match(r'([a-zA-Z0-9]+\.)+[a-zA-Z0-9]{2,}', domain)) @login_required -def request_vhost(request): +def request_vhost(request: HttpRequest) -> HttpResponse: user = logged_in_user(request) attrs = user_attrs(user) is_group = 'callinkOid' in attrs @@ -137,7 +140,9 @@ def request_vhost(request): else: return redirect(reverse('request_vhost_success')) else: - form = VirtualHostForm(is_group, initial={'requested_subdomain': user + '.berkeley.edu'}) + # Unsupported left operand type for + ("None") because form might not have been instantiated at this point... + # but this doesn't matter because of if-else clause + form = VirtualHostForm(is_group, initial={'requested_subdomain': user + '.berkeley.edu'}) # type: ignore group_url = f'https://www.ocf.berkeley.edu/~{user}/' @@ -156,7 +161,7 @@ def request_vhost(request): ) -def request_vhost_success(request): +def request_vhost_success(request: HttpRequest) -> HttpResponse: return render( request, 'account/vhost/success.html', @@ -231,7 +236,7 @@ class VirtualHostForm(Form): max_length=1024, ) - def __init__(self, is_group=True, *args, **kwargs): + def __init__(self, is_group: bool = True, *args: Any, **kwargs: Any) -> None: super(Form, self).__init__(*args, **kwargs) # It's pretty derpy that we have to set the labels here, but we can't @@ -266,7 +271,7 @@ def __init__(self, is_group=True, *args, **kwargs): max_length=64, ) - def clean_requested_subdomain(self): + def clean_requested_subdomain(self) -> str: requested_subdomain = self.cleaned_data['requested_subdomain'].lower().strip() if self.cleaned_data['requested_own_domain']: @@ -291,7 +296,7 @@ def clean_requested_subdomain(self): return requested_subdomain - def clean_your_email(self): + def clean_your_email(self) -> str: your_email = self.cleaned_data['your_email'] if not valid_email(your_email): raise forms.ValidationError( diff --git a/ocfweb/account/vhost_mail.py b/ocfweb/account/vhost_mail.py index 2ca1b9f9f..52fe35fcd 100644 --- a/ocfweb/account/vhost_mail.py +++ b/ocfweb/account/vhost_mail.py @@ -3,10 +3,19 @@ import re from contextlib import contextmanager from textwrap import dedent +from typing import Any +from typing import Collection +from typing import Dict +from typing import Generator +from typing import NoReturn +from typing import Optional +from typing import Tuple from django.conf import settings from django.contrib import messages +from django.http import HttpRequest from django.http import HttpResponse +from django.http import HttpResponseRedirect from django.shortcuts import redirect from django.shortcuts import render from django.urls import reverse @@ -22,7 +31,6 @@ from ocfweb.component.errors import ResponseException from ocfweb.component.session import logged_in_user - EXAMPLE_CSV = dedent("""\ president,john.doe@berkeley.edu officers,john.doe@berkeley.edu jane.doe@berkeley.edu @@ -42,7 +50,7 @@ class InvalidEmailError(ValueError): @login_required @group_account_required -def vhost_mail(request): +def vhost_mail(request: HttpRequest) -> HttpResponse: user = logged_in_user(request) vhosts = [] @@ -69,12 +77,15 @@ def vhost_mail(request): @login_required @group_account_required @require_POST -def vhost_mail_update(request): +def vhost_mail_update(request: HttpRequest) -> HttpResponseRedirect: user = logged_in_user(request) # All requests are required to have these action = _get_action(request) - addr_name, addr_domain, addr_vhost = _get_addr(request, user, 'addr', required=True) + # _get_addr may return None, but never with this particular call + addr_info = _get_addr(request, user, 'addr', required=True) + assert addr_info is not None + addr_name, addr_domain, addr_vhost = addr_info addr = (addr_name or '') + '@' + addr_domain # These fields are optional; some might be None @@ -137,7 +148,7 @@ def vhost_mail_update(request): @login_required @group_account_required -def vhost_mail_csv_export(request, domain): +def vhost_mail_csv_export(request: HttpRequest, domain: str) -> HttpResponse: user = logged_in_user(request) vhost = _get_vhost(user, domain) if not vhost: @@ -161,7 +172,7 @@ def vhost_mail_csv_export(request, domain): @login_required @group_account_required @require_POST -def vhost_mail_csv_import(request, domain): +def vhost_mail_csv_import(request: HttpRequest, domain: str) -> HttpResponseRedirect: user = logged_in_user(request) vhost = _get_vhost(user, domain) if not vhost: @@ -204,7 +215,7 @@ def vhost_mail_csv_import(request, domain): return _redirect_back() -def _write_csv(addresses): +def _write_csv(addresses: Generator[Any, None, None]) -> Any: """Turn a collection of vhost forwarding addresses into a CSV string for user download.""" buf = io.StringIO() @@ -218,14 +229,14 @@ def _write_csv(addresses): return buf.getvalue() -def _parse_csv(request, domain): +def _parse_csv(request: HttpRequest, domain: str) -> Dict[str, Any]: """Parse, validate, and return addresses from the file uploaded with the CSV upload button/form.""" - csv_file = request.FILES.get('csv_file') + csv_file: Any = request.FILES.get('csv_file') if not csv_file: _error(request, 'Missing CSV file!') - addresses = {} + addresses: Dict[str, Collection[Any]] = {} try: with io.TextIOWrapper(csv_file, encoding='utf-8') as f: reader = csv.reader(f) @@ -247,12 +258,12 @@ def _parse_csv(request, domain): except ValueError as e: _error(request, 'Error parsing CSV: row {}: {}'.format(i + 1, e)) except UnicodeDecodeError as e: - _error(f'Uploaded file is not valid UTF-8 encoded: "{e}"') + _error(request, f'Uploaded file is not valid UTF-8 encoded: "{e}"') return addresses -def _parse_csv_forward_addrs(string): +def _parse_csv_forward_addrs(string: str) -> Collection[Any]: """Parse and validate emails from a commas-and-whitespace separated list string.""" # Allow any combination of whitespace and , as separators @@ -269,24 +280,26 @@ def _parse_csv_forward_addrs(string): return frozenset(to_addrs) -def _error(request, msg): +def _error(request: HttpRequest, msg: str) -> NoReturn: messages.add_message(request, messages.ERROR, msg) raise ResponseException(_redirect_back()) -def _redirect_back(): +def _redirect_back() -> Any: return redirect(reverse('vhost_mail')) -def _get_action(request): +def _get_action(request: HttpRequest) -> Any: action = request.POST.get('action') if action not in {'add', 'update', 'delete'}: _error(request, f'Invalid action: "{action}"') else: return action + return None + -def _parse_addr(addr, allow_wildcard=False): +def _parse_addr(addr: str, allow_wildcard: bool = False) -> Optional[Tuple[str, str]]: """Safely parse an email, returning first component and domain.""" m = re.match( ( @@ -302,8 +315,10 @@ def _parse_addr(addr, allow_wildcard=False): if '.' in domain: return name, domain + return None -def _get_addr(request, user, field, required=True): + +def _get_addr(request: HttpRequest, user: Any, field: str, required: bool = True) -> Optional[Tuple[Any, Any, Any]]: original = request.POST.get(field) if original is not None: addr = original.strip() @@ -322,8 +337,10 @@ def _get_addr(request, user, field, required=True): elif required: _error(request, 'You must provide an address!') + return None + -def _get_forward_to(request): +def _get_forward_to(request: HttpRequest) -> Optional[Collection[Any]]: forward_to = request.POST.get('forward_to') if forward_to is None: @@ -346,7 +363,7 @@ def _get_forward_to(request): return frozenset(parsed_addrs) -def _get_password(request, addr_name): +def _get_password(request: HttpRequest, addr_name: Optional[str]) -> Any: # If addr_name is None, then this is a wildcard address, and those can't # have passwords. if addr_name is None: @@ -365,21 +382,21 @@ def _get_password(request, addr_name): return crypt_password(password) -def _get_vhost(user, domain): +def _get_vhost(user: Any, domain: str) -> Any: vhosts = vhosts_for_user(user) for vhost in vhosts: if vhost.domain == domain: return vhost -def _find_addr(c, vhost, addr): +def _find_addr(c: Any, vhost: Any, addr: str) -> Any: for addr_obj in vhost.get_forwarding_addresses(c): if addr_obj.address == addr: return addr_obj @contextmanager -def _txn(**kwargs): +def _txn(**kwargs: Any) -> Generator[Any, None, None]: with get_connection( user=settings.OCFMAIL_USER, password=settings.OCFMAIL_PASSWORD, diff --git a/ocfweb/announcements/announcements.py b/ocfweb/announcements/announcements.py index fd747fef3..df18f7c0e 100644 --- a/ocfweb/announcements/announcements.py +++ b/ocfweb/announcements/announcements.py @@ -1,10 +1,14 @@ from collections import namedtuple from datetime import date -from datetime import datetime +from datetime import datetime as original_datetime from datetime import time +from typing import Any +from typing import Callable from typing import Tuple from cached_property import cached_property +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from django.templatetags.static import static from django.urls import reverse @@ -18,24 +22,24 @@ class Announcement(namedtuple('Announcement', ('title', 'date', 'path', 'render'))): @cached_property - def link(self): + def link(self) -> str: return reverse(self.route_name) @cached_property - def route_name(self): + def route_name(self) -> str: return f'{self.path}-announcement' @cached_property - def datetime(self): + def datetime(self) -> original_datetime: """This is pretty silly, but Django humanize needs a datetime.""" return timezone.make_aware( - datetime.combine(self.date, time()), + original_datetime.combine(self.date, time()), timezone.get_default_timezone(), ) -def announcement(title, date, path): - def wrapper(fn): +def announcement(title: str, date: date, path: str) -> Callable[[Any], Any]: + def wrapper(fn: Callable[..., Any]) -> Callable[..., Any]: global announcements announcements += ( Announcement( @@ -49,7 +53,7 @@ def wrapper(fn): return wrapper -def index(request): +def index(request: HttpRequest) -> HttpResponse: return render( request, 'announcements/index.html', @@ -71,7 +75,7 @@ def index(request): date(2016, 5, 12), 'ocf-eff-alliance', ) -def eff_alliance(title, request): +def eff_alliance(title: str, request: HttpRequest) -> HttpResponse: return render( request, 'announcements/2016-05-12-ocf-eff-alliance.html', @@ -86,7 +90,7 @@ def eff_alliance(title, request): date(2016, 4, 1), 'renaming-ocf', ) -def renaming_announcement(title, request): +def renaming_announcement(title: str, request: HttpRequest) -> HttpResponse: return render( request, 'announcements/2016-04-01-renaming.html', @@ -107,7 +111,7 @@ def renaming_announcement(title, request): date(2016, 2, 9), 'printing', ) -def printing_announcement(title, request): +def printing_announcement(title: str, request: HttpRequest) -> HttpResponse: return render( request, 'announcements/2016-02-09-printing.html', @@ -122,7 +126,7 @@ def printing_announcement(title, request): date(2017, 3, 1), 'hpc-survey', ) -def hpc_survey(title, request): +def hpc_survey(title: str, request: HttpRequest) -> HttpResponse: return render( request, 'announcements/2017-03-01-hpc-survey.html', @@ -137,7 +141,7 @@ def hpc_survey(title, request): date(2017, 3, 20), 'hiring-2017', ) -def hiring_2017(title, request): +def hiring_2017(title: str, request: HttpRequest) -> HttpResponse: return render( request, 'announcements/2017-03-20-hiring.html', @@ -152,7 +156,7 @@ def hiring_2017(title, request): date(2018, 10, 30), 'hiring-2018', ) -def hiring_2018(title, request): +def hiring_2018(title: str, request: HttpRequest) -> HttpResponse: return render( request, 'announcements/2018-10-30-hiring.html', diff --git a/ocfweb/api/announce.py b/ocfweb/api/announce.py index 6e9cf36f0..bcf93ecae 100644 --- a/ocfweb/api/announce.py +++ b/ocfweb/api/announce.py @@ -1,9 +1,10 @@ +from django.http import HttpRequest from django.http import JsonResponse from ocfweb.component.blog import get_blog_posts as real_get_blog_posts -def get_blog_posts(request): +def get_blog_posts(request: HttpRequest) -> JsonResponse: return JsonResponse( [item._asdict() for item in real_get_blog_posts()], safe=False, diff --git a/ocfweb/api/hours.py b/ocfweb/api/hours.py index e49a226f6..2876ca901 100644 --- a/ocfweb/api/hours.py +++ b/ocfweb/api/hours.py @@ -1,6 +1,8 @@ from datetime import time from json import JSONEncoder +from typing import Any +from django.http import HttpRequest from django.http import JsonResponse from ocflib.lab.hours import Hour from ocflib.lab.hours import HoursListing @@ -10,7 +12,7 @@ class JSONHoursEncoder(JSONEncoder): - def default(self, obj): + def default(self, obj: Any) -> Any: if isinstance(obj, HoursListing): return obj.__dict__ elif isinstance(obj, Hour): @@ -22,11 +24,11 @@ def default(self, obj): @periodic(60) -def get_hours_listing(): +def get_hours_listing() -> HoursListing: return read_hours_listing() -def get_hours_today(request): +def get_hours_today(request: HttpRequest) -> JsonResponse: return JsonResponse( get_hours_listing().hours_on_date(), encoder=JSONHoursEncoder, diff --git a/ocfweb/api/lab.py b/ocfweb/api/lab.py index dbb3978aa..f2d9c2faf 100644 --- a/ocfweb/api/lab.py +++ b/ocfweb/api/lab.py @@ -1,3 +1,8 @@ +from typing import Any +from typing import List +from typing import Set + +from django.http import HttpRequest from django.http import JsonResponse from ocflib.infra.hosts import hostname_from_domain from ocflib.lab.stats import get_connection @@ -8,12 +13,12 @@ @cache() -def _list_public_desktops(): +def _list_public_desktops() -> List[Any]: return list_desktops(public_only=True) @periodic(5) -def _get_desktops_in_use(): +def _get_desktops_in_use() -> Set[Any]: """List which desktops are currently in use.""" # https://github.com/ocf/ocflib/blob/90f9268a89ac9d53c089ab819c1aa95bdc38823d/ocflib/lab/ocfstats.sql#L70 @@ -27,7 +32,7 @@ def _get_desktops_in_use(): return {hostname_from_domain(session['host']) for session in c} -def desktop_usage(request): +def desktop_usage(request: HttpRequest) -> JsonResponse: public_desktops = _list_public_desktops() desktops_in_use = _get_desktops_in_use() diff --git a/ocfweb/api/session_tracking.py b/ocfweb/api/session_tracking.py index 8169dc5cf..05bd46852 100644 --- a/ocfweb/api/session_tracking.py +++ b/ocfweb/api/session_tracking.py @@ -2,8 +2,11 @@ from enum import Enum from functools import partial from ipaddress import ip_address +from typing import Any +from typing import Dict from django.conf import settings +from django.http import HttpRequest from django.http import HttpResponse from django.http import HttpResponseBadRequest from django.views.decorators.csrf import csrf_exempt @@ -28,7 +31,7 @@ @require_POST @csrf_exempt -def log_session(request): +def log_session(request: HttpRequest) -> HttpResponse: """Primary API endpoint for session tracking. Desktops have a cronjob that calls this endpoint: https://git.io/vpIKX @@ -63,7 +66,7 @@ def log_session(request): return HttpResponseBadRequest(e) -def _new_session(host, user): +def _new_session(host: str, user: str) -> None: """Register new session in when a user logs into a desktop.""" _close_sessions(host) @@ -75,7 +78,7 @@ def _new_session(host, user): ) -def _session_exists(host, user): +def _session_exists(host: str, user: str) -> bool: """Returns whether an open session already exists for a given host and user.""" with get_connection() as c: @@ -87,7 +90,7 @@ def _session_exists(host, user): return c.fetchone()['count'] > 0 -def _refresh_session(host, user): +def _refresh_session(host: str, user: str) -> None: """Keep a session around if the user is still logged in.""" with get_connection() as c: @@ -97,7 +100,7 @@ def _refresh_session(host, user): ) -def _close_sessions(host): +def _close_sessions(host: str) -> None: """Close all sessions for a particular host.""" with get_connection() as c: @@ -108,7 +111,7 @@ def _close_sessions(host): @cache(600) -def _get_desktops(): +def _get_desktops() -> Dict[Any, Any]: """Return IPv4 and 6 address to fqdn mapping for OCF desktops from LDAP.""" desktops = {} diff --git a/ocfweb/api/shorturls.py b/ocfweb/api/shorturls.py index 70ad232d6..58ba1af7e 100644 --- a/ocfweb/api/shorturls.py +++ b/ocfweb/api/shorturls.py @@ -1,10 +1,14 @@ +from typing import Any +from typing import Union + +from django.http import HttpRequest from django.http import HttpResponseNotFound from django.http import HttpResponseRedirect from ocflib.misc.shorturls import get_connection from ocflib.misc.shorturls import get_shorturl -def bounce_shorturl(request, slug): +def bounce_shorturl(request: HttpRequest, slug: Any) -> Union[HttpResponseRedirect, HttpResponseNotFound]: if slug: with get_connection() as ctx: target = get_shorturl(ctx, slug) diff --git a/ocfweb/api/staff_hours.py b/ocfweb/api/staff_hours.py index f8186d0b4..d6af0b1ba 100644 --- a/ocfweb/api/staff_hours.py +++ b/ocfweb/api/staff_hours.py @@ -1,9 +1,10 @@ +from django.http import HttpRequest from django.http import JsonResponse from ocfweb.main.staff_hours import get_staff_hours as real_get_staff_hours -def get_staff_hours(request): +def get_staff_hours(request: HttpRequest) -> JsonResponse: return JsonResponse( [item._asdict() for item in real_get_staff_hours()], safe=False, diff --git a/ocfweb/auth.py b/ocfweb/auth.py index 22cf62b2b..bc24e1a16 100644 --- a/ocfweb/auth.py +++ b/ocfweb/auth.py @@ -1,7 +1,11 @@ # TODO: move this file into ocfweb.component.session? +from typing import Any +from typing import Callable +from typing import Optional from urllib.parse import urlencode from django.contrib.auth import REDIRECT_FIELD_NAME +from django.http import HttpRequest from django.http import HttpResponseRedirect from django.shortcuts import render from django.urls import reverse @@ -11,8 +15,8 @@ from ocfweb.component.session import logged_in_user -def login_required(function): - def _decorator(request, *args, **kwargs): +def login_required(function: Callable[..., Any]) -> Callable[..., Any]: + def _decorator(request: HttpRequest, *args: Any, **kwargs: Any) -> Any: if is_logged_in(request): return function(request, *args, **kwargs) @@ -22,10 +26,10 @@ def _decorator(request, *args, **kwargs): return _decorator -def group_account_required(function): - def _decorator(request, *args, **kwargs): +def group_account_required(function: Callable[..., Any]) -> Callable[..., Any]: + def _decorator(request: HttpRequest, *args: Any, **kwargs: Any) -> Any: try: - user = logged_in_user(request) + user: Optional[str] = logged_in_user(request) except KeyError: user = None @@ -41,13 +45,13 @@ def _decorator(request, *args, **kwargs): return _decorator -def calnet_required(fn): +def calnet_required(fn: Callable[..., Any]) -> Callable[..., Any]: """Decorator for views that require CalNet auth Checks if "calnet_uid" is in the request.session dictionary. If the value is not a valid uid, the user is rediected to CalNet login view. """ - def wrapper(request, *args, **kwargs): + def wrapper(request: HttpRequest, *args: Any, **kwargs: Any) -> Any: calnet_uid = request.session.get('calnet_uid') if calnet_uid: return fn(request, *args, **kwargs) diff --git a/ocfweb/bin/run_periodic_functions.py b/ocfweb/bin/run_periodic_functions.py index 423be749f..18638c531 100755 --- a/ocfweb/bin/run_periodic_functions.py +++ b/ocfweb/bin/run_periodic_functions.py @@ -12,6 +12,8 @@ from argparse import ArgumentParser from textwrap import dedent from traceback import format_exc +from typing import Any +from typing import Optional from django.conf import settings from ocflib.misc.mail import send_problem_report @@ -21,17 +23,16 @@ from ocfweb.caching import periodic_functions - _logger = logging.getLogger(__name__) logging.basicConfig(level=logging.INFO) # seconds to pause worker after encountering an error DELAY_ON_ERROR_MIN = 30 DELAY_ON_ERROR_MAX = 1800 # 30 minutes -delay_on_error = DELAY_ON_ERROR_MIN +delay_on_error: float = DELAY_ON_ERROR_MIN -def run_periodic_functions(): +def run_periodic_functions() -> None: global delay_on_error # First, import urls so that views are imported, decorators are run, and @@ -99,7 +100,7 @@ def run_periodic_functions(): delay_on_error = max(DELAY_ON_ERROR_MIN, delay_on_error / 2) -def main(argv=None): +def main(argv: Optional[Any] = None) -> int: os.environ['DJANGO_SETTINGS_MODULE'] = 'ocfweb.settings' parser = ArgumentParser(description='Run ocfweb periodic functions') @@ -121,6 +122,8 @@ def main(argv=None): run_periodic_functions() time.sleep(1) + return 0 + if __name__ == '__main__': sys.exit(main()) diff --git a/ocfweb/caching.py b/ocfweb/caching.py index 1aba885e0..21310b92d 100644 --- a/ocfweb/caching.py +++ b/ocfweb/caching.py @@ -4,6 +4,13 @@ from collections import namedtuple from datetime import datetime from itertools import chain +from typing import Any +from typing import Callable +from typing import Dict +from typing import Hashable +from typing import Iterable +from typing import Optional +from typing import Tuple from cached_property import cached_property from django.conf import settings @@ -11,11 +18,10 @@ from ocfweb.environment import ocfweb_version - _logger = logging.getLogger(__name__) -def cache_lookup(key): +def cache_lookup(key: Hashable) -> Any: """Look up a key in the cache, raising KeyError if it's a miss.""" # The "get" method returns `None` both for cached values of `None`, # and keys which aren't in the cache. @@ -23,7 +29,7 @@ def cache_lookup(key): # The recommended workaround is using a sentinel as a default # return value for when a key is missing. This allows us to still # cache functions which return None. - cache_miss_sentinel = {} + cache_miss_sentinel: Dict[Any, Any] = {} retval = django_cache.get(key, cache_miss_sentinel) is_hit = retval is not cache_miss_sentinel @@ -35,7 +41,9 @@ def cache_lookup(key): return retval -def cache_lookup_with_fallback(key, fallback, ttl=None, force_miss=False): +def cache_lookup_with_fallback( + key: Hashable, fallback: Callable[[], Any], ttl: Optional[int] = None, force_miss: bool = False, +) -> Any: """Look up a key in the cache, falling back to a function if it's a miss. We first check if the key is in the cache, and if so, return it. If not, we @@ -69,7 +77,7 @@ def cache_lookup_with_fallback(key, fallback, ttl=None, force_miss=False): return result -def cache(ttl=None): +def cache(ttl: Optional[int] = None) -> Callable[[Callable[..., Any]], Callable[..., Any]]: """Caching function decorator, with an optional ttl. The optional ttl (in seconds) specifies how long cache entries should live. @@ -94,8 +102,8 @@ def my_deterministic_function(a, b, c): def my_changing_function(a, b, c): .... """ - def outer(fn): - def inner(*args, **kwargs): + def outer(fn: Callable[..., Any]) -> Callable[..., Any]: + def inner(*args: Any, **kwargs: Any) -> Any: return cache_lookup_with_fallback( _make_function_call_key(fn, args, kwargs), lambda: fn(*args, **kwargs), @@ -105,7 +113,7 @@ def inner(*args, **kwargs): return outer -def _make_key(key): +def _make_key(key: Iterable[Any]) -> Tuple[Any, ...]: """Return a key suitable for caching. The returned key prepends a version tag so that we don't share the cache @@ -122,7 +130,7 @@ def _make_key(key): ) -def _make_function_call_key(fn, args, kwargs): +def _make_function_call_key(fn: Callable[..., Any], args: Iterable[Any], kwargs: Dict[Any, Any]) -> Tuple[Any, ...]: """Return a key for a function call. The key will eventually be converted to a string and used as a cache key. @@ -152,21 +160,25 @@ class PeriodicFunction( ), ): - def __hash__(self): + def __hash__(self) -> int: return hash(self.function_call_key) - def __eq__(self, other): + def __eq__(self, other: object) -> bool: + # Mypy note: It is recommended for __eq__ to work with arbitrary objects. + if not isinstance(other, PeriodicFunction): + return NotImplemented + return self.function_call_key == other.function_call_key - def __str__(self): + def __str__(self) -> str: return f'PeriodicFunction({self.function_call_key})' @cached_property - def function_call_key(self): + def function_call_key(self) -> Tuple[Any, ...]: """Return the function's cache key.""" return _make_function_call_key(self.function, (), {}) - def function_with_timestamp(self): + def function_with_timestamp(self) -> Tuple[datetime, Any]: """Return a tuple (timestamp, result). This is the value we actually store in the cache; the benefit is that @@ -176,7 +188,7 @@ def function_with_timestamp(self): """ return (datetime.now(), self.function()) - def last_update(self): + def last_update(self) -> Any: """Return the timestamp of the last update of this function. If the function has never been updated, returns None.""" @@ -186,7 +198,7 @@ def last_update(self): except KeyError: return None - def seconds_since_last_update(self): + def seconds_since_last_update(self) -> float: """Return the number of seconds since the last update. If we've never updated, we return the number of seconds since @@ -195,7 +207,7 @@ def seconds_since_last_update(self): last_update = self.last_update() or datetime.fromtimestamp(0) return (datetime.now() - last_update).total_seconds() - def result(self, **kwargs): + def result(self, **kwargs: Any) -> Any: """Return the result of this periodic function. In most cases, we can read it from the cache and so it is nearly @@ -212,7 +224,7 @@ def result(self, **kwargs): ) return result - def update(self): + def update(self) -> Any: """Run this periodic function and cache the result.""" cache_lookup_with_fallback( self.function_call_key, @@ -222,7 +234,7 @@ def update(self): ) -def periodic(period, ttl=None): +def periodic(period: float, ttl: Optional[float] = None) -> Callable[[Callable[..., Any]], Any]: """Caching function decorator for functions which desire TTL-based caching. Using this decorator on a function registers it as a "periodic function", @@ -259,7 +271,7 @@ def get_blog_posts(): elif ttl is None: ttl = period * 2 - def outer(fn): + def outer(fn: Callable[..., Any]) -> Any: pf = PeriodicFunction( function=fn, period=period, diff --git a/ocfweb/component/blog.py b/ocfweb/component/blog.py index c69e11d92..a618de856 100644 --- a/ocfweb/component/blog.py +++ b/ocfweb/component/blog.py @@ -1,4 +1,7 @@ from collections import namedtuple +from typing import Any +from typing import Dict +from typing import List from xml.etree import ElementTree as etree import dateutil.parser @@ -8,7 +11,6 @@ from ocfweb.caching import periodic - _namespaces = {'atom': 'http://www.w3.org/2005/Atom'} @@ -28,32 +30,34 @@ class Post( ): @cached_property - def datetime(self): + def datetime(self) -> bool: return self.published @classmethod - def from_element(cls, element): - def grab_attr(attr): - el = element + def from_element(cls: Any, element: Any) -> Any: + def grab_attr(attr: str) -> str: + el: Any = element for part in attr.split('_'): el = el.find('atom:' + part, namespaces=_namespaces) return el.text - attrs = { + attrs: Dict[str, Any] = { attr: grab_attr(attr) for attr in cls._fields } attrs['updated'] = dateutil.parser.parse(attrs['updated']) attrs['published'] = dateutil.parser.parse(attrs['published']) - attrs['link'] = element.find( + # Fix builtin function being typed as returning an int on error, which has no get + el_find: Any = element.find( './/atom:link[@type="text/html"]', namespaces=_namespaces, - ).get('href') + ) + attrs['link'] = el_find.get('href') return cls(**attrs) @periodic(60) -def get_blog_posts(): +def get_blog_posts() -> List[Any]: """Parse the beautiful OCF status blog atom feed into a list of Posts. Unfortunately Blogger is hella flakey so we use it inside a loop and fail diff --git a/ocfweb/component/errors.py b/ocfweb/component/errors.py index 67beea6d1..a26cc4018 100644 --- a/ocfweb/component/errors.py +++ b/ocfweb/component/errors.py @@ -1,4 +1,7 @@ +from requests import models + + class ResponseException(Exception): - def __init__(self, response): + def __init__(self, response: models.Response) -> None: self.response = response diff --git a/ocfweb/component/forms.py b/ocfweb/component/forms.py index 1a12835ba..fc601a9f9 100644 --- a/ocfweb/component/forms.py +++ b/ocfweb/component/forms.py @@ -1,3 +1,6 @@ +from typing import Any +from typing import Callable + from django import forms from django.core.exceptions import ValidationError @@ -9,7 +12,7 @@ class Form(forms.Form): required_css_class = 'required' -def wrap_validator(validator): +def wrap_validator(validator: Callable[..., Any]) -> Callable[..., None]: """Wraps a validator which raises some kind of Exception, and instead returns a Django ValidationError with the same message. @@ -21,7 +24,7 @@ def wrap_validator(validator): >>> validator('ocf') ValidationError: Username is reserved """ - def wrapped_validator(*args, **kwargs): + def wrapped_validator(*args: Any, **kwargs: Any) -> None: try: validator(*args, **kwargs) except Exception as ex: diff --git a/ocfweb/component/graph.py b/ocfweb/component/graph.py index 3d93ed270..00ba9c06f 100644 --- a/ocfweb/component/graph.py +++ b/ocfweb/component/graph.py @@ -3,25 +3,33 @@ from datetime import date from datetime import datetime from datetime import timedelta +from typing import Any +from typing import Callable +from typing import Optional +from typing import Tuple +from django.http import HttpRequest from django.http import HttpResponse from django.shortcuts import redirect from django.urls import reverse from matplotlib.backends.backend_agg import FigureCanvasAgg - +from matplotlib.figure import Figure MIN_DAYS = 1 MAX_DAYS = 365 * 5 DEFAULT_DAYS = 14 -def current_start_end(): +def current_start_end() -> Tuple[date, date]: """Return current default start and end date.""" end = date.today() return end - timedelta(days=DEFAULT_DAYS), end -def canonical_graph(hot_path=None, default_start_end=current_start_end): +def canonical_graph( + hot_path: Optional[Callable[..., Any]] = None, + default_start_end: Callable[..., Tuple[date, date]] = current_start_end, +) -> Callable[..., Any]: """Decorator to make graphs with a start_day and end_day. It does three primary things: @@ -42,9 +50,9 @@ def canonical_graph(hot_path=None, default_start_end=current_start_end): :param default_start_end: optional, function to get current start/end date (default: current_start_end) """ - def decorator(fn): - def wrapper(request): - def _day_from_params(param, default): + def decorator(fn: Callable[[Any, date, date], Any]) -> Callable[[Any], Any]: + def wrapper(request: HttpRequest) -> Any: + def _day_from_params(param: str, default: date) -> date: try: return datetime.strptime(request.GET.get(param, ''), '%Y-%m-%d').date() except ValueError: @@ -85,7 +93,7 @@ def _day_from_params(param, default): return decorator -def plot_to_image_bytes(fig, format='svg', **kwargs): +def plot_to_image_bytes(fig: Figure, format: str = 'svg', **kwargs: Any) -> bytes: """Return bytes representing the plot image.""" buf = io.BytesIO() FigureCanvasAgg(fig).print_figure(buf, format=format, **kwargs) diff --git a/ocfweb/component/lab_status.py b/ocfweb/component/lab_status.py index cc91cd691..790f70a0d 100644 --- a/ocfweb/component/lab_status.py +++ b/ocfweb/component/lab_status.py @@ -14,7 +14,7 @@ @periodic(60, ttl=86400) -def get_lab_status(): +def get_lab_status() -> LabStatus: """Get the front page banner message from the default location.""" with open('/etc/ocf/lab_status.yaml') as f: tree = yaml.safe_load(f) diff --git a/ocfweb/component/markdown.py b/ocfweb/component/markdown.py index 487c432c8..02685e332 100644 --- a/ocfweb/component/markdown.py +++ b/ocfweb/component/markdown.py @@ -1,4 +1,10 @@ import re +from typing import Any +from typing import List +from typing import Match +from typing import Set +from typing import Tuple +from typing import TYPE_CHECKING import mistune from django.urls import reverse @@ -14,24 +20,39 @@ # tags of a format like: [[!meta title="Backups"]] META_REGEX = re.compile(r'\[\[!meta ([a-z]+)="([^"]*)"\]\]') +# Make mypy play nicely with mixins https://github.com/python/mypy/issues/5837 +# TODO: issue has been resolved in mypy, patch when next version of mypy gets released +# More info: https://github.com/python/mypy/pull/7860 -class HtmlCommentsLexerMixin: + +class MixinBase: + def __init__(self, rules: Any, default_rules: Any) -> None: + self.rules = rules + self.default_rules = default_rules + + +_Base: Any = object +if TYPE_CHECKING: + _Base = MixinBase + + +class HtmlCommentsLexerMixin(_Base): """Strip HTML comments as entire blocks or inside lines.""" - def enable_html_comments(self): + def enable_html_comments(self) -> None: self.rules.html_comment = re.compile( r'^', ) self.default_rules.insert(0, 'html_comment') - def output_html_comment(self, m): + def output_html_comment(self, m: Match[Any]) -> str: return '' - def parse_html_comment(self, m): + def parse_html_comment(self, m: Match[Any]) -> None: pass -class BackslashLineBreakLexerMixin: +class BackslashLineBreakLexerMixin(_Base): """Convert lines that end in a backslash into a simple line break. This follows GitHub-flavored Markdown on backslashes at the end of lines @@ -50,22 +71,22 @@ class BackslashLineBreakLexerMixin: with a line break """ - def enable_backslash_line_breaks(self): + def enable_backslash_line_breaks(self) -> None: self.rules.backslash_line_break = re.compile( '^\\\\\n', ) self.default_rules.insert(0, 'backslash_line_break') - def output_backslash_line_break(self, m): + def output_backslash_line_break(self, m: Match[Any]) -> str: return '
' -class CodeRendererMixin: +class CodeRendererMixin(_Base): """Render highlighted code.""" # TODO: don't use inline styles; see http://pygments.org/docs/formatters/ html_formatter = HtmlFormatter(noclasses=True) - def block_code(self, code, lang): + def block_code(self, code: str, lang: str) -> str: try: if lang: lexer = get_lexer_by_name(lang, stripall=True) @@ -77,7 +98,7 @@ def block_code(self, code, lang): return highlight(code, lexer, CodeRendererMixin.html_formatter) -class DjangoLinkInlineLexerMixin: +class DjangoLinkInlineLexerMixin(_Base): """Turn special Markdown link syntax into Django links. In Django templates, we can use `url` tags, such as: @@ -95,7 +116,7 @@ class DjangoLinkInlineLexerMixin: split_words = re.compile(r'((?:\S|\\ )+)') - def enable_django_links(self): + def enable_django_links(self) -> None: self.rules.django_link = re.compile( r'^\[\[(?!\!)' r'([\s\S]+?)' @@ -106,10 +127,10 @@ def enable_django_links(self): ) self.default_rules.insert(0, 'django_link') - def output_django_link(self, m): + def output_django_link(self, m: Match[Any]) -> str: text, target, fragment = m.group(1), m.group(2), m.group(3) - def href(link, fragment): + def href(link: str, fragment: str) -> str: if fragment: return link + '#' + fragment return link @@ -123,7 +144,7 @@ def href(link, fragment): ) -class HeaderRendererMixin: +class HeaderRendererMixin(_Base): """Mixin to render headers with auto-generated IDs (or provided IDs). If headers are written as usual, they'll be given automatically-generated @@ -142,14 +163,14 @@ class HeaderRendererMixin: rendering a document and read afterwards. """ - def reset_toc(self): - self.toc = [] - self.toc_ids = set() + def reset_toc(self) -> None: + self.toc: List[Any] = [] + self.toc_ids: Set[Any] = set() - def get_toc(self): + def get_toc(self) -> List[Any]: return self.toc - def header(self, text, level, raw=None): + def header(self, text: str, level: int, raw: None = None) -> str: custom_id_match = re.match(r'^(.*?)\s+{([a-z0-9\-_]+)}\s*$', text) if custom_id_match: text = custom_id_match.group(1) @@ -220,12 +241,12 @@ class OcfMarkdownBlockLexer( ) -def markdown(text): +def markdown(text: str) -> mistune.Markdown: _renderer.reset_toc() return _markdown(text) -def text_and_meta(f): +def text_and_meta(f: Any) -> Tuple[str, Any]: """Return tuple (text, meta dict) for the given file. Meta tags are stripped from the Markdown source, but the Markdown is @@ -234,7 +255,7 @@ def text_and_meta(f): text = f.read() meta = {} - def repl(match): + def repl(match: Match[Any]) -> str: meta[match.group(1)] = match.group(2) return '' @@ -243,7 +264,7 @@ def repl(match): @cache() -def markdown_and_toc(text): +def markdown_and_toc(text: str) -> Tuple[Any, Any]: """Return tuple (html, toc) for the given text.""" html = markdown(text) return html, _renderer.get_toc() diff --git a/ocfweb/component/session.py b/ocfweb/component/session.py index eda3271a7..fd3b571fe 100644 --- a/ocfweb/component/session.py +++ b/ocfweb/component/session.py @@ -1,23 +1,27 @@ +from typing import Any +from typing import Optional + +from django.http import HttpRequest from ocflib.account.validators import user_exists -def is_logged_in(request): +def is_logged_in(request: HttpRequest) -> bool: """Return whether a user is logged in.""" return bool(logged_in_user(request)) -def logged_in_user(request): +def logged_in_user(request: HttpRequest) -> Optional[Any]: """Return logged in user, or raise KeyError.""" return request.session.get('ocf_user') -def login(request, user): +def login(request: HttpRequest, user: str) -> None: """Log in a user. Doesn't do any kind of password validation (obviously).""" assert user_exists(user) request.session['ocf_user'] = user -def logout(request): +def logout(request: HttpRequest) -> bool: """Log out the user. Return True if a user was logged out, False otherwise.""" try: del request.session['ocf_user'] diff --git a/ocfweb/context_processors.py b/ocfweb/context_processors.py index bceb2753c..f01a18882 100644 --- a/ocfweb/context_processors.py +++ b/ocfweb/context_processors.py @@ -1,6 +1,10 @@ import re from ipaddress import ip_address +from typing import Any +from typing import Dict +from typing import Generator +from django.http import HttpRequest from django.urls import reverse from ipware import get_client_ip from ocflib.account.search import user_is_group @@ -12,7 +16,7 @@ from ocfweb.environment import ocfweb_version -def get_base_css_classes(request): +def get_base_css_classes(request: HttpRequest) -> Generator[str, None, None]: if request.resolver_match and request.resolver_match.url_name: page_class = 'page-' + request.resolver_match.url_name yield page_class @@ -22,7 +26,7 @@ def get_base_css_classes(request): yield page_class -def ocf_template_processor(request): +def ocf_template_processor(request: HttpRequest) -> Dict[str, Any]: hours_listing = get_hours_listing() real_ip, _ = get_client_ip(request) user = logged_in_user(request) diff --git a/ocfweb/docs/doc.py b/ocfweb/docs/doc.py index 11e957679..0e78522f4 100644 --- a/ocfweb/docs/doc.py +++ b/ocfweb/docs/doc.py @@ -6,7 +6,7 @@ class Document(namedtuple('Document', ['name', 'title', 'render'])): @cached_property - def category(self): + def category(self) -> str: """Return full category path of the document. For example, "/" or "/staff/backend/". @@ -14,7 +14,7 @@ def category(self): return self.name.rsplit('/', 1)[0] + '/' @cached_property - def category_for_sidebar(self): + def category_for_sidebar(self) -> str: """Return the category to show similar pages for in the sidebar. If this page isn't at the root category, we just return this page's @@ -29,7 +29,7 @@ def category_for_sidebar(self): return self.category @cached_property - def edit_url(self): + def edit_url(self) -> str: """Return a GitHub edit URL for this page.""" return ( 'https://github.com/ocf/ocfweb/edit/master/ocfweb/docs/docs' + @@ -38,7 +38,7 @@ def edit_url(self): ) @cached_property - def history_url(self): + def history_url(self) -> str: """Return a GitHub history URL for this page.""" return ( 'https://github.com/ocf/ocfweb/commits/master/ocfweb/docs/docs' + diff --git a/ocfweb/docs/markdown_based.py b/ocfweb/docs/markdown_based.py index 383217b95..976bd053b 100644 --- a/ocfweb/docs/markdown_based.py +++ b/ocfweb/docs/markdown_based.py @@ -15,19 +15,25 @@ import os from functools import partial from pathlib import Path +from typing import Any +from typing import Dict +from typing import Generator from django.conf import settings +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocfweb.component.markdown import markdown_and_toc from ocfweb.component.markdown import text_and_meta from ocfweb.docs.doc import Document - DOCS_DIR = Path(__file__).parent.joinpath('docs') -def render_markdown_doc(path, meta, text, doc, request): +def render_markdown_doc( + path: Path, meta: Dict[str, Any], text: str, doc: Document, request: HttpRequest, +) -> HttpResponse: # Reload markdown docs if in development if settings.DEBUG: @@ -48,7 +54,7 @@ def render_markdown_doc(path, meta, text, doc, request): ) -def get_markdown_docs(): +def get_markdown_docs() -> Generator[Document, None, None]: for path in DOCS_DIR.glob('**/*.md'): name, _ = os.path.splitext(str(path.relative_to(DOCS_DIR))) diff --git a/ocfweb/docs/templatetags/docs.py b/ocfweb/docs/templatetags/docs.py index bcf56ee85..5d3821125 100644 --- a/ocfweb/docs/templatetags/docs.py +++ b/ocfweb/docs/templatetags/docs.py @@ -1,6 +1,11 @@ import re from collections import namedtuple from operator import attrgetter +from typing import Any +from typing import Collection +from typing import Dict +from typing import List +from typing import Optional from django import template from django.utils.html import strip_tags @@ -11,7 +16,7 @@ class Node(namedtuple('Node', ['path', 'title', 'children'])): @property - def url_path(self): + def url_path(self) -> str: return self.path.lstrip('/').rstrip('/') @@ -19,14 +24,19 @@ def url_path(self): @register.inclusion_tag('docs/partials/doc-tree.html') -def doc_tree(root='/', suppress_root=True, cur_path=None, exclude='$^'): +def doc_tree( + root: str = '/', + suppress_root: bool = True, + cur_path: Optional[str] = None, + exclude: Any = '$^', +) -> Dict[str, Any]: # root is expected to be like '/' or '/services/' or '/services/web/' assert root.startswith('/') assert root.endswith('/') exclude = re.compile(exclude) - def _make_tree(root): + def _make_tree(root: str) -> Node: path = root[:-1] doc = DOCS.get(path) return Node( @@ -54,10 +64,10 @@ def _make_tree(root): @register.inclusion_tag('docs/partials/doc-toc.html') -def doc_toc(toc, collapsible=False): +def doc_toc(toc: Collection[Any], collapsible: bool = False) -> Dict[str, Any]: if len(toc) > 3: # heuristic to avoid dumb tables of contents - levels = list(sorted({entry[0] for entry in toc})) - cur = levels[0] + levels: List[Any] = list(sorted({entry[0] for entry in toc})) + cur: int = levels[0] html = '
    ' diff --git a/ocfweb/docs/urls.py b/ocfweb/docs/urls.py index 47f510ea0..cef9f0936 100644 --- a/ocfweb/docs/urls.py +++ b/ocfweb/docs/urls.py @@ -3,6 +3,9 @@ from django.conf.urls import url from django.http import Http404 +from django.http import HttpRequest +from django.http import HttpResponse +from django.http import HttpResponseRedirect from django.shortcuts import redirect from django.urls import reverse @@ -17,7 +20,6 @@ from ocfweb.docs.views.officers import officers from ocfweb.docs.views.servers import servers - DOCS = { doc.name: doc for doc in chain( @@ -40,7 +42,7 @@ } -def render_doc(request, doc_name): +def render_doc(request: HttpRequest, doc_name: str) -> HttpResponse: """Render a document given a request.""" doc = DOCS['/' + doc_name] if not doc: @@ -48,13 +50,13 @@ def render_doc(request, doc_name): return doc.render(doc, request) -def send_redirect(request, redir_src): +def send_redirect(request: HttpRequest, redir_src: str) -> HttpResponseRedirect: """Send a redirect to the actual document given the redirecting page.""" redir_dest = REDIRECTS['/' + redir_src] return redirect(reverse('doc', args=(redir_dest,)), permanent=True) -def doc_name(doc_name): +def doc_name(doc_name: str) -> str: # we can't actually deal with escaping into a regex, so we just use a whitelist assert re.match(r'^/[a-zA-Z0-9\-/]+$', doc_name), 'Bad document name: ' + doc_name return doc_name[1:].replace('-', '\\-') diff --git a/ocfweb/docs/views/account_policies.py b/ocfweb/docs/views/account_policies.py index fe5f5d615..41e8b8533 100644 --- a/ocfweb/docs/views/account_policies.py +++ b/ocfweb/docs/views/account_policies.py @@ -1,7 +1,11 @@ +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render +from ocfweb.docs.doc import Document -def account_policies(doc, request): + +def account_policies(doc: Document, request: HttpRequest) -> HttpResponse: return render( request, 'docs/account_policies.html', diff --git a/ocfweb/docs/views/buster_upgrade.py b/ocfweb/docs/views/buster_upgrade.py index db531b656..ca834d8bc 100644 --- a/ocfweb/docs/views/buster_upgrade.py +++ b/ocfweb/docs/views/buster_upgrade.py @@ -1,9 +1,15 @@ from collections import namedtuple +from typing import Any +from typing import Optional +from typing import Tuple +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocflib.misc.validators import host_exists from ocfweb.caching import cache +from ocfweb.docs.doc import Document from ocfweb.docs.views.servers import Host @@ -22,7 +28,7 @@ class ThingToUpgrade( UPGRADED = 3 @classmethod - def from_hostname(cls, hostname, status=NEEDS_UPGRADE, comments=None): + def from_hostname(cls: Any, hostname: str, status: int = NEEDS_UPGRADE, comments: Optional[str] = None) -> Any: has_dev = host_exists('dev-' + hostname + '.ocf.berkeley.edu') return cls( host=Host.from_ldap(hostname), @@ -33,7 +39,7 @@ def from_hostname(cls, hostname, status=NEEDS_UPGRADE, comments=None): @cache() -def _get_servers(): +def _get_servers() -> Tuple[Any, ...]: return ( # login servers ThingToUpgrade.from_hostname( @@ -194,7 +200,7 @@ def _get_servers(): ) -def buster_upgrade(doc, request): +def buster_upgrade(doc: Document, request: HttpRequest) -> HttpResponse: return render( request, 'docs/buster_upgrade.html', diff --git a/ocfweb/docs/views/commands.py b/ocfweb/docs/views/commands.py index 0e8d0dee2..d9280b5f6 100644 --- a/ocfweb/docs/views/commands.py +++ b/ocfweb/docs/views/commands.py @@ -1,8 +1,12 @@ from typing import NamedTuple from typing import Optional +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render +from ocfweb.docs.doc import Document + class Command(NamedTuple): name: str @@ -80,7 +84,7 @@ class Command(NamedTuple): ] -def commands(doc, request): +def commands(doc: Document, request: HttpRequest) -> HttpResponse: return render( request, 'docs/commands.html', diff --git a/ocfweb/docs/views/hosting_badges.py b/ocfweb/docs/views/hosting_badges.py index 814336f31..4c25c7c44 100644 --- a/ocfweb/docs/views/hosting_badges.py +++ b/ocfweb/docs/views/hosting_badges.py @@ -1,8 +1,12 @@ +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from django.urls import reverse +from ocfweb.docs.doc import Document -def hosting_badges(doc, request): + +def hosting_badges(doc: Document, request: HttpRequest) -> HttpResponse: badges = [ (name, request.build_absolute_uri(reverse('hosting-logo', args=(name,)))) for name in [ diff --git a/ocfweb/docs/views/index.py b/ocfweb/docs/views/index.py index 6a47971b2..53e10dca5 100644 --- a/ocfweb/docs/views/index.py +++ b/ocfweb/docs/views/index.py @@ -1,7 +1,9 @@ +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render -def docs_index(request): +def docs_index(request: HttpRequest) -> HttpResponse: return render( request, 'docs/index.html', diff --git a/ocfweb/docs/views/lab.py b/ocfweb/docs/views/lab.py index 559995720..4ae0b8849 100644 --- a/ocfweb/docs/views/lab.py +++ b/ocfweb/docs/views/lab.py @@ -1,12 +1,15 @@ from datetime import date from datetime import timedelta +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocfweb.api.hours import get_hours_listing +from ocfweb.docs.doc import Document -def lab(doc, request): +def lab(doc: Document, request: HttpRequest) -> HttpResponse: hours_listing = get_hours_listing() return render( request, diff --git a/ocfweb/docs/views/officers.py b/ocfweb/docs/views/officers.py index cf5582cca..b011f504e 100644 --- a/ocfweb/docs/views/officers.py +++ b/ocfweb/docs/views/officers.py @@ -1,17 +1,30 @@ import math from collections import namedtuple from datetime import date - +from typing import Any +from typing import Callable +from typing import List +from typing import Optional +from typing import Tuple +from typing import Union + +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocflib.account.search import user_attrs from ocfweb import caching - _Term = namedtuple('_Term', ['name', 'gms', 'sms', 'dgms', 'dsms']) -def Term(name, gms, sms, dgms=None, dsms=None): +def Term( + name: str, + gms: List[Any], + sms: List[Any], + dgms: Optional[List[Any]] = None, + dsms: Optional[List[Any]] = None, +) -> _Term: gms = list(map(Officer.from_uid_or_info, gms)) sms = list(map(Officer.from_uid_or_info, sms)) dgms = list(map(Officer.from_uid_or_info, dgms or [])) @@ -22,7 +35,7 @@ def Term(name, gms, sms, dgms=None, dsms=None): class Officer(namedtuple('Officer', ['uid', 'name', 'start', 'end', 'acting'])): @classmethod - def from_uid_or_info(cls, uid_or_info): + def from_uid_or_info(cls: Callable[..., Any], uid_or_info: Union[Tuple[Any, ...], str]) -> Any: if isinstance(uid_or_info, tuple): if len(uid_or_info) == 3: uid, start, end = uid_or_info @@ -40,10 +53,10 @@ def from_uid_or_info(cls, uid_or_info): return cls(uid=uid, name=name, start=start, end=end, acting=acting) @property - def full_term(self): + def full_term(self) -> bool: return self.start is None and self.end is None - def __str__(self): + def __str__(self) -> str: s = f'{self.name} <{self.uid}>' if self.acting: if self.end is not None and self.end < date(2016, 11, 14): @@ -77,7 +90,7 @@ def __str__(self): # This function makes approximately five million LDAP queries, so it's # important that these terms aren't loaded at import time. @caching.periodic(math.inf) -def _bod_terms(): +def _bod_terms() -> List[Any]: return [ Term( 'Spring 1989', @@ -229,7 +242,7 @@ def _bod_terms(): ] -def officers(doc, request): +def officers(doc: Any, request: HttpRequest) -> HttpResponse: terms = _bod_terms() return render( request, diff --git a/ocfweb/docs/views/servers.py b/ocfweb/docs/views/servers.py index 571458d39..01f39985b 100644 --- a/ocfweb/docs/views/servers.py +++ b/ocfweb/docs/views/servers.py @@ -1,13 +1,20 @@ import os from collections import namedtuple +from typing import Any +from typing import Dict +from typing import List +from typing import Tuple import dns.resolver import requests from cached_property import cached_property +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocflib.infra.hosts import hosts_by_filter from ocfweb.caching import cache +from ocfweb.docs.doc import Document PUPPETDB_URL = 'https://puppetdb:8081/pdb/query/v4' PUPPET_CERT_DIR = '/etc/ocfweb/puppet-certs' @@ -15,7 +22,7 @@ class Host(namedtuple('Host', ['hostname', 'type', 'description', 'children'])): @classmethod - def from_ldap(cls, hostname, type='vm', children=()): + def from_ldap(cls: Any, hostname: str, type: str = 'vm', children: Any = ()) -> Any: host = hosts_by_filter(f'(cn={hostname})') if 'description' in host: description, = host['description'] @@ -29,21 +36,22 @@ def from_ldap(cls, hostname, type='vm', children=()): ) @cached_property - def ipv4(self): + def ipv4(self) -> str: try: - return str(dns.resolver.query(self.hostname, 'A')[0]) + # for this and ipv6 below: dns.resolver.query is not typed but is within a package. + return str(dns.resolver.query(self.hostname, 'A')[0]) # type: ignore except dns.resolver.NXDOMAIN: return 'No IPv4 Address' @cached_property - def ipv6(self): + def ipv6(self) -> str: try: - return str(dns.resolver.query(self.hostname, 'AAAA')[0]) + return str(dns.resolver.query(self.hostname, 'AAAA')[0]) # type: ignore except (dns.resolver.NoAnswer, dns.resolver.NXDOMAIN): return 'No IPv6 address' @cached_property - def english_type(self): + def english_type(self) -> str: return { 'desktop': 'Desktop', 'hypervisor': 'Hypervisor', @@ -57,10 +65,10 @@ def english_type(self): }[self.type] @cached_property - def has_munin(self): + def has_munin(self) -> bool: return self.type in ('hypervisor', 'vm', 'server', 'desktop') - def __key(self): + def __key(self) -> Tuple[Any, str, str]: """Key function used for comparison.""" ranking = { 'hypervisor': 1, @@ -70,11 +78,11 @@ def __key(self): default = 3 return (ranking.get(self.type, default), self.type, self.hostname) - def __lt__(self, other_host): + def __lt__(self: Any, other_host: Any) -> bool: return self.__key() < other_host.__key() -def is_hidden(host): +def is_hidden(host: Dict[Any, Any]) -> bool: return host['cn'][0].startswith('hozer-') or host['cn'][0].startswith('dev-') @@ -82,7 +90,7 @@ def is_hidden(host): PQL_IS_HYPERVISOR = 'resources[certname] { type = "Class" and title = "Ocf_kvm" }' -def query_puppet(query): +def query_puppet(query: str) -> Dict[Any, Any]: """Accepts a PQL query, returns a parsed json result.""" r = requests.get( PUPPETDB_URL, @@ -96,12 +104,12 @@ def query_puppet(query): return r.json() if r.status_code == 200 else None -def format_query_output(item): +def format_query_output(item: Dict[Any, Any]) -> Tuple[Any, Any]: """Converts an item of a puppet query to tuple(hostname, query_value).""" return item['certname'].split('.')[0], item.get('value') -def ldap_to_host(item): +def ldap_to_host(item: Any) -> Tuple[Any, Any]: """Accepts an ldap output item, returns tuple(hostname, host_object).""" description = item.get('description', [''])[0] hostname = item['cn'][0] @@ -109,12 +117,12 @@ def ldap_to_host(item): @cache() -def get_hosts(): +def get_hosts() -> List[Any]: ldap_output = hosts_by_filter('(|(type=server)(type=desktop)(type=printer))') - servers = dict(ldap_to_host(item) for item in ldap_output if not is_hidden(item)) + servers: Dict[Any, Any] = dict(ldap_to_host(item) for item in ldap_output if not is_hidden(item)) - hypervisors_hostnames = dict(format_query_output(item) for item in query_puppet(PQL_IS_HYPERVISOR)) - all_children = dict(format_query_output(item) for item in query_puppet(PQL_GET_VMS)) + hypervisors_hostnames: Dict[Any, Any] = dict(format_query_output(item) for item in query_puppet(PQL_IS_HYPERVISOR)) + all_children: Dict[Any, Any] = dict(format_query_output(item) for item in query_puppet(PQL_GET_VMS)) hostnames_seen = { # These are manually added later, with the correct type @@ -160,7 +168,7 @@ def get_hosts(): return sorted(servers_to_display) -def servers(doc, request): +def servers(doc: Document, request: HttpRequest) -> HttpResponse: return render( request, 'docs/servers.html', diff --git a/ocfweb/environment.py b/ocfweb/environment.py index be97b4434..f153e2ac4 100644 --- a/ocfweb/environment.py +++ b/ocfweb/environment.py @@ -4,7 +4,7 @@ @lru_cache() -def ocfweb_version(): +def ocfweb_version() -> str: """Return string representing ocfweb version. In dev, returns 'dev'. In prod, returns a version diff --git a/ocfweb/login/calnet.py b/ocfweb/login/calnet.py index 86ba8e450..110009c71 100644 --- a/ocfweb/login/calnet.py +++ b/ocfweb/login/calnet.py @@ -1,14 +1,18 @@ +from typing import Any +from typing import Optional +from typing import Union from urllib.parse import urlencode from urllib.parse import urljoin import ocflib.ucb.cas as cas from django.contrib.auth import REDIRECT_FIELD_NAME +from django.http import HttpRequest from django.http import HttpResponse from django.http import HttpResponseForbidden from django.http import HttpResponseRedirect -def _service_url(request, next_page): +def _service_url(request: HttpRequest, next_page: Optional[str]) -> str: protocol = ('http://', 'https://')[request.is_secure()] host = request.get_host() service = protocol + host + request.path @@ -18,7 +22,7 @@ def _service_url(request, next_page): return url -def _redirect_url(request): +def _redirect_url(request: HttpRequest) -> Optional[Any]: """ Redirects to referring page """ next_page = request.META.get('HTTP_REFERER') prefix = ('http://', 'https://')[request.is_secure()] + request.get_host() @@ -27,7 +31,7 @@ def _redirect_url(request): return next_page -def _login_url(service): +def _login_url(service: str) -> str: params = { 'service': service, 'renew': 'true', @@ -37,7 +41,7 @@ def _login_url(service): ) -def _logout_url(request, next_page=None): +def _logout_url(request: HttpRequest, next_page: Optional[str] = None) -> str: url = urljoin(cas.CAS_URL, 'logout') if next_page: protocol = ('http://', 'https://')[request.is_secure()] @@ -46,7 +50,7 @@ def _logout_url(request, next_page=None): return url -def _next_page_response(next_page): +def _next_page_response(next_page: Optional[str]) -> Union[HttpResponse, HttpResponseRedirect]: if next_page: return HttpResponseRedirect(next_page) else: @@ -55,7 +59,7 @@ def _next_page_response(next_page): ) -def login(request, next_page=None): +def login(request: HttpRequest, next_page: Any = None) -> HttpResponse: next_page = request.GET.get(REDIRECT_FIELD_NAME) if not next_page: next_page = _redirect_url(request) @@ -77,7 +81,7 @@ def login(request, next_page=None): return HttpResponseRedirect(_login_url(service)) -def logout(request, next_page=None): +def logout(request: HttpRequest, next_page: Optional[str] = None) -> Union[HttpResponse, HttpResponseRedirect]: if 'calnet_uid' in request.session: del request.session['calnet_uid'] if not next_page: diff --git a/ocfweb/login/ocf.py b/ocfweb/login/ocf.py index f6f0950b8..406e9837e 100644 --- a/ocfweb/login/ocf.py +++ b/ocfweb/login/ocf.py @@ -1,8 +1,14 @@ import re +from typing import Any +from typing import Match +from typing import Optional +from typing import Union import ocflib.account.utils as utils import ocflib.account.validators as validators from django import forms +from django.http import HttpRequest +from django.http import HttpResponse from django.http import HttpResponseRedirect from django.shortcuts import render from django.urls import reverse @@ -15,7 +21,7 @@ from ocfweb.component.session import logout as session_logout -def _valid_return_path(return_to): +def _valid_return_path(return_to: str) -> Optional[Match[Any]]: """Make sure this is a valid relative path to prevent redirect attacks.""" return re.match( '^/[^/]', @@ -23,7 +29,7 @@ def _valid_return_path(return_to): ) -def login(request): +def login(request: HttpRequest) -> Union[HttpResponseRedirect, HttpResponse]: error = None return_to = request.GET.get('next') @@ -69,7 +75,7 @@ def login(request): @login_required -def logout(request): +def logout(request: HttpRequest) -> Union[HttpResponseRedirect, HttpResponse]: return_to = request.GET.get('next') if return_to and _valid_return_path(return_to): request.session['login_return_path'] = return_to @@ -93,7 +99,7 @@ def logout(request): ) -def redirect_back(request): +def redirect_back(request: HttpRequest) -> HttpResponseRedirect: """Return the user to the page they were trying to access, or the home page if we don't know what they were trying to access. """ @@ -116,6 +122,6 @@ class LoginForm(Form): max_length=64, ) - def clean_username(self): + def clean_username(self) -> str: username = self.cleaned_data.get('username', '') return username.strip().lower() diff --git a/ocfweb/main/favicon.py b/ocfweb/main/favicon.py index e8e59ab8d..da0a7bba7 100644 --- a/ocfweb/main/favicon.py +++ b/ocfweb/main/favicon.py @@ -1,10 +1,11 @@ from os.path import dirname from os.path import join +from django.http import HttpRequest from django.http import HttpResponse -def favicon(request): +def favicon(request: HttpRequest) -> HttpResponse: """favicon.ico must be served from the root for legacy reasons.""" with open(join(dirname(dirname(__file__)), 'static', 'img', 'favicon', 'favicon.ico'), 'rb') as f: return HttpResponse( diff --git a/ocfweb/main/home.py b/ocfweb/main/home.py index df305484a..619902af5 100644 --- a/ocfweb/main/home.py +++ b/ocfweb/main/home.py @@ -2,6 +2,8 @@ from datetime import timedelta from operator import attrgetter +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocflib.lab.staff_hours import get_staff_hours_soonest_first @@ -13,11 +15,11 @@ @periodic(60, ttl=86400) -def get_staff_hours(): +def get_staff_hours() -> str: return get_staff_hours_soonest_first()[:2] -def home(request): +def home(request: HttpRequest) -> HttpResponse: hours_listing = get_hours_listing() hours = [ ( diff --git a/ocfweb/main/hosting_logos.py b/ocfweb/main/hosting_logos.py index 7a5c8ab5d..f2acc042c 100644 --- a/ocfweb/main/hosting_logos.py +++ b/ocfweb/main/hosting_logos.py @@ -4,14 +4,18 @@ from os.path import isfile from os.path import join from os.path import realpath +from typing import Optional +from typing import Tuple +from typing import Union from django.http import Http404 +from django.http import HttpRequest from django.http import HttpResponse +from django.http import HttpResponseRedirect from django.shortcuts import redirect from ocfweb.caching import cache - # images not in PNG that we now redirect to PNG versions LEGACY_IMAGES = [ 'berknow150x40.jpg', @@ -36,7 +40,7 @@ @cache() -def get_image(image): +def get_image(image: str) -> Tuple[bytes, Optional[str]]: match = re.match(r'^[a-z0-9_\-]+\.(png|svg)$', image) if not match: raise Http404() @@ -55,7 +59,7 @@ def get_image(image): return f.read(), content_type -def hosting_logo(request, image): +def hosting_logo(request: HttpRequest, image: str) -> Union[HttpResponse, HttpResponseRedirect]: """Hosting logos must be served from the root since they are linked by student group websites.""" # legacy images diff --git a/ocfweb/main/robots.py b/ocfweb/main/robots.py index b795a2015..2eb785fe1 100644 --- a/ocfweb/main/robots.py +++ b/ocfweb/main/robots.py @@ -1,10 +1,11 @@ from textwrap import dedent from django.conf import settings +from django.http import HttpRequest from django.http import HttpResponse -def robots_dot_txt(request): +def robots_dot_txt(request: HttpRequest) -> HttpResponse: """Serve /robots.txt file.""" if settings.DEBUG: resp = """\ diff --git a/ocfweb/main/security.py b/ocfweb/main/security.py index 6bd260722..8a83e7749 100644 --- a/ocfweb/main/security.py +++ b/ocfweb/main/security.py @@ -1,3 +1,4 @@ +from django.http import HttpRequest from django.http import HttpResponse SECURITY_TXT = """\ @@ -6,6 +7,6 @@ """ -def security_dot_txt(request): +def security_dot_txt(request: HttpRequest) -> HttpResponse: """Serve the security.txt file.""" return HttpResponse(SECURITY_TXT, content_type='text/plain') diff --git a/ocfweb/main/staff_hours.py b/ocfweb/main/staff_hours.py index 915827ca9..83aa1e17e 100644 --- a/ocfweb/main/staff_hours.py +++ b/ocfweb/main/staff_hours.py @@ -1,5 +1,9 @@ import time +from typing import Any +from typing import List +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocflib.lab.staff_hours import get_staff_hours as real_get_staff_hours @@ -8,11 +12,11 @@ @periodic(60, ttl=86400) -def get_staff_hours(): +def get_staff_hours() -> List[Any]: return real_get_staff_hours() -def staff_hours(request): +def staff_hours(request: HttpRequest) -> HttpResponse: return render( request, 'main/staff-hours.html', diff --git a/ocfweb/main/templatetags/staff_hours.py b/ocfweb/main/templatetags/staff_hours.py index 08180c2b4..2cf2e688e 100644 --- a/ocfweb/main/templatetags/staff_hours.py +++ b/ocfweb/main/templatetags/staff_hours.py @@ -1,8 +1,10 @@ +from typing import Any + from django import template register = template.Library() @register.filter -def gravatar(staffer, size): +def gravatar(staffer: Any, size: int) -> str: return staffer.gravatar(size) diff --git a/ocfweb/middleware/errors.py b/ocfweb/middleware/errors.py index 0af46d4ec..9b0b7c750 100644 --- a/ocfweb/middleware/errors.py +++ b/ocfweb/middleware/errors.py @@ -2,21 +2,25 @@ from pprint import pformat from textwrap import dedent from traceback import format_exc +from typing import Any +from typing import Callable +from typing import Dict +from typing import Iterable from django.conf import settings +from django.http import HttpRequest from django.http.response import Http404 from ocflib.misc.mail import send_problem_report from ocfweb.component.errors import ResponseException - SENSITIVE_WSGI_CONTEXT = frozenset(( 'HTTP_COOKIE', 'CSRF_COOKIE', )) -def sanitize(msg): +def sanitize(msg: str) -> str: """Attempt to sanitize out known-bad patterns.""" # Remove any dictionary references with "encrypted_password", e.g. lines like: # {'some_key': ..., 'encrypted_password': b'asdf', 'some_other_key': ...} @@ -24,7 +28,7 @@ def sanitize(msg): return msg -def sanitize_wsgi_context(headers): +def sanitize_wsgi_context(headers: Iterable[Any]) -> Dict[Any, Any]: """Attempt to sanitize out known-bad WSGI context keys.""" headers = dict(headers) for key in SENSITIVE_WSGI_CONTEXT: @@ -35,13 +39,13 @@ def sanitize_wsgi_context(headers): class OcflibErrorMiddleware: - def __init__(self, get_response): + def __init__(self, get_response: Callable[..., Any]) -> None: self.get_response = get_response - def __call__(self, request): + def __call__(self, request: HttpRequest) -> Any: return self.get_response(request) - def process_exception(self, request, exception): + def process_exception(self, request: HttpRequest, exception: Exception) -> Any: if isinstance(exception, ResponseException): return exception.response diff --git a/ocfweb/settings.py b/ocfweb/settings.py index 47720cf40..44e3e9974 100644 --- a/ocfweb/settings.py +++ b/ocfweb/settings.py @@ -69,7 +69,7 @@ class InvalidReferenceInTemplate(str): exception. """ - def __mod__(self, ref): + def __mod__(self, ref: Any) -> str: raise TemplateSyntaxError(f'Invalid reference in template: {ref}') diff --git a/ocfweb/stats/accounts.py b/ocfweb/stats/accounts.py index 2d05097e0..b59972460 100644 --- a/ocfweb/stats/accounts.py +++ b/ocfweb/stats/accounts.py @@ -2,7 +2,14 @@ from collections import defaultdict from datetime import date from datetime import timedelta +from typing import Any +from typing import DefaultDict +from typing import Dict +from typing import Hashable +from typing import List +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocflib.infra.ldap import ldap_ocf from ocflib.infra.ldap import OCF_LDAP_PEOPLE @@ -10,7 +17,7 @@ from ocfweb.caching import cache -def stats_accounts(request): +def stats_accounts(request: HttpRequest) -> HttpResponse: account_data = _get_account_stats() return render( request, @@ -38,7 +45,7 @@ def stats_accounts(request): @cache(ttl=600) -def _get_account_stats(): +def _get_account_stats() -> Dict[str, List[Any]]: with ldap_ocf() as c: c.search(OCF_LDAP_PEOPLE, '(cn=*)', attributes=['creationTime', 'uidNumber', 'callinkOid']) response = c.response @@ -48,8 +55,8 @@ def _get_account_stats(): start_date = date(1995, 8, 21) last_creation_time = start_date sorted_accounts = sorted(response, key=lambda record: record['attributes']['uidNumber']) - counts = defaultdict(int) - group_counts = defaultdict(int) + counts: DefaultDict[Hashable, int] = defaultdict(int) + group_counts: DefaultDict[Hashable, int] = defaultdict(int) for account in sorted_accounts: creation_time = account['attributes'].get('creationTime', None) diff --git a/ocfweb/stats/daily_graph.py b/ocfweb/stats/daily_graph.py index dac3594ed..ad4013813 100644 --- a/ocfweb/stats/daily_graph.py +++ b/ocfweb/stats/daily_graph.py @@ -2,7 +2,11 @@ from datetime import date from datetime import datetime from datetime import timedelta +from typing import Any +from typing import Optional +from typing import Tuple +from django.http import HttpRequest from django.http import HttpResponse from django.shortcuts import redirect from django.urls import reverse @@ -15,13 +19,12 @@ from ocfweb.caching import periodic from ocfweb.component.graph import plot_to_image_bytes - # Binomial-shaped weights for moving average AVERAGE_WEIGHTS = tuple(zip(range(-2, 3), (n / 16 for n in (1, 4, 6, 4, 1)))) @periodic(60) -def _daily_graph_image(day=None): +def _daily_graph_image(day: Optional[date] = None) -> HttpResponse: if not day: day = date.today() @@ -31,7 +34,7 @@ def _daily_graph_image(day=None): ) -def daily_graph_image(request): +def daily_graph_image(request: HttpRequest) -> Any: try: day = datetime.strptime(request.GET.get('date', ''), '%Y-%m-%d').date() except ValueError: @@ -52,7 +55,7 @@ def daily_graph_image(request): return _daily_graph_image(day=day) -def get_open_close(day): +def get_open_close(day: date) -> Tuple[datetime, datetime]: """Return datetime objects representing open and close for a day rounded down to the hour. @@ -82,14 +85,14 @@ def get_open_close(day): # TODO: caching; we can cache for a long time if it's a day that's already happened -def get_daily_plot(day): +def get_daily_plot(day: date) -> Figure: """Return matplotlib plot representing a day's plot.""" start, end = get_open_close(day) desktops = list_desktops(public_only=True) profiles = UtilizationProfile.from_hostnames(desktops, start, end).values() desks_count = len(desktops) - now = datetime.now() + now: Any = datetime.now() latest = min(end, now) minute = timedelta(minutes=1) times = [start + i * minute for i in range((latest - start) // minute + 1)] @@ -104,7 +107,7 @@ def get_daily_plot(day): sums.append(in_use) # Do a weighted moving average to smooth out the data - processed = [0] * len(sums) + processed = [0.0] * len(sums) for i in range(len(sums)): for delta_i, weight in AVERAGE_WEIGHTS: m = i if (i + delta_i < 0 or i + delta_i >= len(sums)) else i + delta_i diff --git a/ocfweb/stats/job_frequency.py b/ocfweb/stats/job_frequency.py index 90fd3a629..f39185051 100644 --- a/ocfweb/stats/job_frequency.py +++ b/ocfweb/stats/job_frequency.py @@ -1,9 +1,12 @@ import urllib.parse from datetime import date from datetime import datetime +from typing import Any +from typing import Optional import numpy as np import ocflib.printing.quota as quota +from django.http import HttpRequest from django.http import HttpResponse from django.shortcuts import redirect from django.urls import reverse @@ -14,13 +17,13 @@ from ocfweb.component.graph import plot_to_image_bytes -def pyday_to_sqlday(pyday): +def pyday_to_sqlday(pyday: int) -> int: """Converting weekday index from python to mysql.""" return (pyday + 1) % 7 + 1 @periodic(1800) -def _jobs_graph_image(day=None): +def _jobs_graph_image(day: Optional[date] = None) -> HttpResponse: if not day: day = date.today() @@ -30,7 +33,7 @@ def _jobs_graph_image(day=None): ) -def daily_jobs_image(request): +def daily_jobs_image(request: HttpRequest) -> Any: try: day = datetime.strptime(request.GET.get('date', ''), '%Y-%m-%d').date() except ValueError: @@ -51,11 +54,11 @@ def daily_jobs_image(request): return _jobs_graph_image(day=day) -def get_jobs_plot(day): +def get_jobs_plot(day: date) -> Figure: """Return matplotlib plot showing the number i-page-job to the day.""" - day_of_week = pyday_to_sqlday(day.weekday()) - day_quota = quota.daily_quota(datetime.combine(day, datetime.min.time())) + day_of_week: int = pyday_to_sqlday(day.weekday()) + day_quota: int = quota.daily_quota(datetime.combine(day, datetime.min.time())) sql_today_freq = ''' SELECT `pages`, SUM(`count`) AS `count` diff --git a/ocfweb/stats/mirrors.py b/ocfweb/stats/mirrors.py index 67a92627c..75e31c0f8 100644 --- a/ocfweb/stats/mirrors.py +++ b/ocfweb/stats/mirrors.py @@ -1,5 +1,9 @@ from datetime import date +from typing import Any +from typing import Tuple +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocflib.lab.stats import bandwidth_by_dist from ocflib.lab.stats import current_semester_start @@ -10,7 +14,7 @@ MIRRORS_EPOCH = date(2017, 1, 1) -def stats_mirrors(request): +def stats_mirrors(request: HttpRequest) -> HttpResponse: semester_total, semester_dists = bandwidth_semester() all_time_total, all_time_dists = bandwidth_all_time() @@ -30,7 +34,7 @@ def stats_mirrors(request): @periodic(86400) -def bandwidth_semester(): +def bandwidth_semester() -> Tuple[Any, Any]: data = bandwidth_by_dist(current_semester_start()) @@ -41,7 +45,7 @@ def bandwidth_semester(): @periodic(86400) -def bandwidth_all_time(): +def bandwidth_all_time() -> Tuple[Any, Any]: data = bandwidth_by_dist(MIRRORS_EPOCH) diff --git a/ocfweb/stats/printing.py b/ocfweb/stats/printing.py index 0c00ea240..7e47282b4 100644 --- a/ocfweb/stats/printing.py +++ b/ocfweb/stats/printing.py @@ -3,7 +3,11 @@ from datetime import date from datetime import timedelta from functools import partial +from typing import Any +from typing import Dict +from typing import List +from django.http import HttpRequest from django.http import HttpResponse from django.shortcuts import render from matplotlib.figure import Figure @@ -15,12 +19,11 @@ from ocfweb.caching import periodic from ocfweb.component.graph import plot_to_image_bytes - ALL_PRINTERS = ('papercut', 'pagefault', 'logjam', 'logjam-old', 'deforestation') ACTIVE_PRINTERS = ('papercut', 'pagefault', 'logjam') -def stats_printing(request): +def stats_printing(request: HttpRequest) -> HttpResponse: return render( request, 'stats/printing.html', @@ -37,7 +40,7 @@ def stats_printing(request): ) -def semester_histogram(request): +def semester_histogram(request: HttpRequest) -> HttpResponse: return HttpResponse( plot_to_image_bytes(_semester_histogram(), format='svg'), content_type='image/svg+xml', @@ -45,7 +48,7 @@ def semester_histogram(request): @periodic(300) -def _semester_histogram(): +def _semester_histogram() -> Figure: with get_connection() as c: c.execute( 'SELECT `user`, `semester` FROM `printed` WHERE `semester` > 0', @@ -66,7 +69,7 @@ def _semester_histogram(): @periodic(3600) -def _toner_changes(): +def _toner_changes() -> List[Any]: return [ ( printer, @@ -76,7 +79,7 @@ def _toner_changes(): ] -def _toner_used_by_printer(printer, cutoff=.05, since=None): +def _toner_used_by_printer(printer: str, cutoff: float = .05, since: date = stats.current_semester_start()) -> float: """Returns toner used for a printer since a given date (by default it returns toner used for this semester). @@ -87,9 +90,6 @@ def _toner_used_by_printer(printer, cutoff=.05, since=None): count diffs that are smaller than a cutoff which empirically seems to be more accurate. """ - if not since: - since = stats.current_semester_start() - with stats.get_connection() as cursor: cursor.execute( ''' @@ -143,7 +143,7 @@ def _toner_used_by_printer(printer, cutoff=.05, since=None): @periodic(120) -def _pages_per_day(): +def _pages_per_day() -> Dict[str, int]: with stats.get_connection() as cursor: cursor.execute(''' SELECT max(value) as value, cast(date as date) as date, printer @@ -155,8 +155,8 @@ def _pages_per_day(): # Resolves the issue of possible missing dates. # defaultdict(lambda: defaultdict(int)) doesn't work due to inability to pickle local objects like lambdas; # this effectively does the same thing as that. - pages_printed = defaultdict(partial(defaultdict, int)) - last_seen = {} + pages_printed: Dict[Any, Any] = defaultdict(partial(defaultdict, int)) + last_seen: Dict[Any, Any] = {} for row in cursor: if row['printer'] in last_seen: @@ -169,7 +169,7 @@ def _pages_per_day(): return pages_printed -def _pages_printed_for_printer(printer, resolution=100): +def _pages_printed_for_printer(printer: str, resolution: int = 100) -> List[Any]: with stats.get_connection() as cursor: cursor.execute( ''' @@ -196,7 +196,7 @@ def _pages_printed_for_printer(printer, resolution=100): @periodic(3600) -def _pages_printed_data(): +def _pages_printed_data() -> List[Any]: return [ { 'name': printer, @@ -207,7 +207,7 @@ def _pages_printed_data(): ] -def pages_printed(request): +def pages_printed(request: HttpRequest) -> HttpResponse: return render( request, 'stats/printing/pages-printed.html', diff --git a/ocfweb/stats/semester_job.py b/ocfweb/stats/semester_job.py index 17dc11e39..8e15b72df 100644 --- a/ocfweb/stats/semester_job.py +++ b/ocfweb/stats/semester_job.py @@ -1,4 +1,10 @@ +from datetime import datetime +from typing import Any +from typing import Sequence +from typing import Sized + import ocflib.printing.quota as quota +from django.http import HttpRequest from django.http import HttpResponse from matplotlib.figure import Figure from matplotlib.ticker import MaxNLocator @@ -11,7 +17,7 @@ @canonical_graph(default_start_end=semester_dates) -def weekday_jobs_image(request, start_day, end_day): +def weekday_jobs_image(request: HttpRequest, start_day: datetime, end_day: datetime) -> HttpResponse: return HttpResponse( plot_to_image_bytes(get_jobs_plot('weekday', start_day, end_day), format='svg'), content_type='image/svg+xml', @@ -19,7 +25,7 @@ def weekday_jobs_image(request, start_day, end_day): @canonical_graph(default_start_end=semester_dates) -def weekend_jobs_image(request, start_day, end_day): +def weekend_jobs_image(request: HttpRequest, start_day: datetime, end_day: datetime) -> HttpResponse: return HttpResponse( plot_to_image_bytes(get_jobs_plot('weekend', start_day, end_day), format='svg'), content_type='image/svg+xml', @@ -58,9 +64,9 @@ def weekend_jobs_image(request, start_day, end_day): def freq_plot( - data, title, - ylab='Number of Jobs Printed', -): + data: Sized, title: Sequence[Any], + ylab: str = 'Number of Jobs Printed', +) -> Figure: """takes in data, title, and ylab and makes a histogram, with the 1:len(data) as the xaxis """ @@ -81,13 +87,15 @@ def freq_plot( return fig -def get_jobs_plot(graph, start_day, end_day): +def get_jobs_plot(graph: str, start_day: datetime, end_day: datetime) -> HttpResponse: """Return matplotlib plot of the number of jobs of different page-jobs""" graph_config = graphs[graph] with quota.get_connection() as cursor: + # Fix mypy error unsupported left operand type for + (Sequence[Any]) + q: Any = graph_config['quota'] cursor.execute( graph_config['query'], - graph_config['quota'] + (start_day, end_day), + q + (start_day, end_day), ) data = cursor.fetchall() diff --git a/ocfweb/stats/session_count.py b/ocfweb/stats/session_count.py index f08f5fed7..fb02b0c1c 100644 --- a/ocfweb/stats/session_count.py +++ b/ocfweb/stats/session_count.py @@ -2,6 +2,7 @@ from datetime import date from datetime import timedelta +from django.http import HttpRequest from django.http import HttpResponse from matplotlib.figure import Figure from ocflib.lab.stats import get_connection @@ -11,28 +12,27 @@ from ocfweb.component.graph import current_start_end from ocfweb.component.graph import plot_to_image_bytes - ONE_DAY = timedelta(days=1) @periodic(60) -def _todays_session_image(): +def _todays_session_image() -> HttpResponse: return _sessions_image(*current_start_end()) @canonical_graph(hot_path=_todays_session_image) -def session_count_image(request, start_day, end_day): +def session_count_image(request: HttpRequest, start_day: date, end_day: date) -> HttpResponse: return _sessions_image(start_day, end_day) -def _sessions_image(start_day, end_day): +def _sessions_image(start_day: date, end_day: date) -> HttpResponse: return HttpResponse( plot_to_image_bytes(get_sessions_plot(start_day, end_day), format='svg'), content_type='image/svg+xml', ) -def get_sessions_plot(start_day, end_day): +def get_sessions_plot(start_day: date, end_day: date) -> Figure: """Return matplotlib plot representing unique sessions between start and end day..""" diff --git a/ocfweb/stats/session_length.py b/ocfweb/stats/session_length.py index c0722b080..a67d1bd57 100644 --- a/ocfweb/stats/session_length.py +++ b/ocfweb/stats/session_length.py @@ -1,7 +1,11 @@ import time from datetime import date +from datetime import datetime from datetime import timedelta +from typing import Any +from typing import Tuple +from django.http import HttpRequest from django.http import HttpResponse from matplotlib.figure import Figure from ocflib.lab.stats import get_connection @@ -10,19 +14,18 @@ from ocfweb.component.graph import canonical_graph from ocfweb.component.graph import plot_to_image_bytes - DEFAULT_DAYS = 90 ONE_DAY = timedelta(days=1) -def current_start_end(): +def current_start_end() -> Tuple[date, date]: """Return current default start and end date.""" end = date.today() return end - timedelta(days=DEFAULT_DAYS), end @periodic(60) -def _todays_session_image(): +def _todays_session_image() -> HttpResponse: return _sessions_image(*current_start_end()) @@ -30,18 +33,18 @@ def _todays_session_image(): hot_path=_todays_session_image, default_start_end=current_start_end, ) -def session_length_image(request, start_day, end_day): +def session_length_image(request: HttpRequest, start_day: datetime, end_day: datetime) -> HttpResponse: return _sessions_image(start_day, end_day) -def _sessions_image(start_day, end_day): +def _sessions_image(start_day: Any, end_day: Any) -> HttpResponse: return HttpResponse( plot_to_image_bytes(get_sessions_plot(start_day, end_day), format='svg'), content_type='image/svg+xml', ) -def get_sessions_plot(start_day, end_day): +def get_sessions_plot(start_day: datetime, end_day: datetime) -> Figure: """Return matplotlib plot representing median session length between start and end day..""" diff --git a/ocfweb/stats/session_stats.py b/ocfweb/stats/session_stats.py index 0f9e66cb3..49244f06f 100644 --- a/ocfweb/stats/session_stats.py +++ b/ocfweb/stats/session_stats.py @@ -1,3 +1,8 @@ +from typing import Any +from typing import List + +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocflib.lab.stats import current_semester_start from ocflib.lab.stats import SESSIONS_EPOCH @@ -8,16 +13,16 @@ @periodic(300) -def top_staff_alltime(): +def top_staff_alltime() -> List[Any]: return real_top_staff_alltime() @periodic(300) -def top_staff_semester(): +def top_staff_semester() -> List[Any]: return real_top_staff_semester() -def session_stats(request): +def session_stats(request: HttpRequest) -> HttpResponse: return render( request, 'stats/session_stats.html', diff --git a/ocfweb/stats/summary.py b/ocfweb/stats/summary.py index 5b82522c8..b5b6c3971 100644 --- a/ocfweb/stats/summary.py +++ b/ocfweb/stats/summary.py @@ -2,7 +2,12 @@ from datetime import date from datetime import datetime from operator import attrgetter +from typing import Any +from typing import Callable +from typing import List +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocflib.lab.stats import current_semester_start from ocflib.lab.stats import list_desktops @@ -20,12 +25,11 @@ from ocfweb.caching import periodic from ocfweb.stats.daily_graph import get_open_close - _logger = logging.getLogger(__name__) @periodic(60) -def desktop_profiles(): +def desktop_profiles() -> List[Any]: open_, close = get_open_close(date.today()) now = datetime.today() @@ -47,34 +51,34 @@ def desktop_profiles(): @periodic(30) -def staff_in_lab(): +def staff_in_lab() -> List[Any]: return real_staff_in_lab() @periodic(300) -def top_staff_alltime(): +def top_staff_alltime() -> List[Any]: return real_top_staff_alltime() @periodic(300) -def top_staff_semester(): +def top_staff_semester() -> List[Any]: return real_top_staff_semester() @periodic(30) -def users_in_lab_count(): +def users_in_lab_count() -> int: return real_users_in_lab_count() @periodic(30) -def staff_in_lab_count(): +def staff_in_lab_count() -> int: return real_staff_in_lab_count() @periodic(60) -def printers(): - def silence(f): - def inner(*args, **kwargs): +def printers() -> List[Any]: + def silence(f: Callable[..., Any]) -> Callable[..., Any]: + def inner(*args: Any, **kwargs: Any) -> Any: try: return f(*args, **kwargs) except (OSError, ValueError) as ex: @@ -88,7 +92,7 @@ def inner(*args, **kwargs): ) -def summary(request): +def summary(request: HttpRequest) -> HttpResponse: return render( request, 'stats/summary.html', diff --git a/ocfweb/stats/templatetags/stats.py b/ocfweb/stats/templatetags/stats.py index 11dbcc2fd..1a61a18f5 100644 --- a/ocfweb/stats/templatetags/stats.py +++ b/ocfweb/stats/templatetags/stats.py @@ -1,4 +1,7 @@ from collections import namedtuple +from typing import Any +from typing import Dict +from typing import Mapping from django import template from django.urls import reverse @@ -9,7 +12,7 @@ @register.inclusion_tag('stats/partials/stats-navbar.html', takes_context=True) -def stats_navbar(context): +def stats_navbar(context: Mapping[Any, Any]) -> Dict[str, Any]: return { 'navbar': [ _NavItem( diff --git a/ocfweb/templatetags/common.py b/ocfweb/templatetags/common.py index a9ce5c743..bb126105e 100644 --- a/ocfweb/templatetags/common.py +++ b/ocfweb/templatetags/common.py @@ -1,4 +1,7 @@ import json as json_ +from typing import Any +from typing import Iterable +from typing import MutableMapping from django import template @@ -6,7 +9,7 @@ @register.filter -def getitem(obj, item): +def getitem(obj: MutableMapping[Any, Any], item: Any) -> Any: """Grab the item from the object. Example usage: @@ -16,13 +19,13 @@ def getitem(obj, item): @register.filter -def sum_values(obj): +def sum_values(obj: Any) -> Any: """Return sum of the object's values.""" return sum(obj.values()) @register.filter -def sort(items): +def sort(items: Iterable[Any]) -> Iterable[Any]: """Sort items. Consider using the built-in `dictsort` filter if you're sorting @@ -35,7 +38,7 @@ def sort(items): @register.filter -def join(items, s): +def join(items: Iterable[Any], s: str) -> str: """Join items (probably of an array). Example usage: @@ -45,5 +48,5 @@ def join(items, s): @register.filter -def json(obj): +def json(obj: object) -> str: return json_.dumps(obj) diff --git a/ocfweb/templatetags/google_maps.py b/ocfweb/templatetags/google_maps.py index 105d0ef5f..b02994c02 100644 --- a/ocfweb/templatetags/google_maps.py +++ b/ocfweb/templatetags/google_maps.py @@ -1,3 +1,5 @@ +from typing import Any +from typing import Dict from urllib.parse import urlencode from uuid import uuid4 @@ -14,7 +16,7 @@ @register.inclusion_tag('partials/google-map.html') -def google_map(width, height, show_info=True): +def google_map(width: float, height: float, show_info: bool = True) -> Dict[str, Any]: return { 'width': width, 'height': height, @@ -27,7 +29,7 @@ def google_map(width, height, show_info=True): @register.inclusion_tag('partials/google-map-static.html') -def google_map_static(width, height): +def google_map_static(width: float, height: float) -> Dict[str, Any]: return { 'url': 'https://maps.googleapis.com/maps/api/staticmap?{}'.format( urlencode({ diff --git a/ocfweb/templatetags/lab_hours.py b/ocfweb/templatetags/lab_hours.py index 3eb25d781..2796cb87b 100644 --- a/ocfweb/templatetags/lab_hours.py +++ b/ocfweb/templatetags/lab_hours.py @@ -1,4 +1,7 @@ from datetime import date +from typing import Any +from typing import Iterable +from typing import Optional from django import template @@ -6,7 +9,7 @@ @register.simple_tag -def lab_hours_holiday(holidays, when=None): +def lab_hours_holiday(holidays: Iterable[Any], when: Optional[date] = None) -> str: if when is None: when = date.today() @@ -17,7 +20,7 @@ def lab_hours_holiday(holidays, when=None): @register.filter -def lab_hours_time(hours): +def lab_hours_time(hours: Optional[Iterable[Any]]) -> str: if hours: return ',\xa0\xa0'.join( # two non-breaking spaces f'{hour.open:%-I:%M%P}–{hour.close:%-I:%M%P}' diff --git a/ocfweb/templatetags/pygments.py b/ocfweb/templatetags/pygments.py index b6f5055dd..42da8b843 100644 --- a/ocfweb/templatetags/pygments.py +++ b/ocfweb/templatetags/pygments.py @@ -1,4 +1,6 @@ from textwrap import dedent +from typing import Any +from typing import Iterable from django import template from pygments import highlight @@ -8,22 +10,14 @@ register = template.Library() -@register.tag -def pygments(parser, token): - _, lang = token.split_contents() - nodelist = parser.parse(('endpygments',)) - parser.delete_first_token() - return PygmentsNode(nodelist, lang) - - class PygmentsNode(template.Node): html_formatter = HtmlFormatter(noclasses=True) - def __init__(self, nodes, lang): + def __init__(self, nodes: Iterable[Any], lang: str) -> None: self.nodes = nodes self.lang = lang - def render(self, context): + def render(self, context: template.Context) -> str: return highlight( dedent( ''.join( @@ -34,3 +28,11 @@ def render(self, context): get_lexer_by_name(self.lang), self.html_formatter, ) + + +@register.tag +def pygments(parser: Any, token: Any) -> PygmentsNode: + _, lang = token.split_contents() + nodelist = parser.parse(('endpygments',)) + parser.delete_first_token() + return PygmentsNode(nodelist, lang) diff --git a/ocfweb/templatetags/ui_components.py b/ocfweb/templatetags/ui_components.py index e3a5358a9..b3b091d7e 100644 --- a/ocfweb/templatetags/ui_components.py +++ b/ocfweb/templatetags/ui_components.py @@ -1,10 +1,12 @@ -from django import template +from typing import Any +from typing import Dict +from django import template register = template.Library() @register.inclusion_tag('partials/progress-bar.html') -def progress_bar(label, value, max): +def progress_bar(label: str, value: int, max: int) -> Dict[str, Any]: """Render a Bootstrap progress bar. :param label: diff --git a/ocfweb/test/periodic.py b/ocfweb/test/periodic.py index 7ba6a823a..654902829 100644 --- a/ocfweb/test/periodic.py +++ b/ocfweb/test/periodic.py @@ -1,11 +1,13 @@ from operator import attrgetter +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import render from ocfweb.caching import periodic_functions -def test_list_periodic_functions(request): +def test_list_periodic_functions(request: HttpRequest) -> HttpResponse: return render( request, 'test/periodic.html', diff --git a/ocfweb/test/session.py b/ocfweb/test/session.py index 95532b6ae..386faf807 100644 --- a/ocfweb/test/session.py +++ b/ocfweb/test/session.py @@ -1,9 +1,10 @@ from os import getpid +from django.http import HttpRequest from django.http import HttpResponse -def test_session(request): +def test_session(request: HttpRequest) -> HttpResponse: request.session.setdefault('n', 0) request.session['n'] += 1 return HttpResponse('pid={} n={}'.format(getpid(), request.session['n'])) diff --git a/ocfweb/tv/main.py b/ocfweb/tv/main.py index 9c06b2a50..2e70e21e0 100644 --- a/ocfweb/tv/main.py +++ b/ocfweb/tv/main.py @@ -1,10 +1,12 @@ +from django.http import HttpRequest +from django.http import HttpResponse from django.shortcuts import redirect from django.shortcuts import render from ocfweb.api.hours import get_hours_listing -def tv_main(request): +def tv_main(request: HttpRequest) -> HttpResponse: return render( request, 'tv/tv.html', @@ -14,5 +16,5 @@ def tv_main(request): ) -def tv_labmap(request): +def tv_labmap(request: HttpRequest) -> HttpResponse: return redirect('https://labmap.ocf.berkeley.edu/') diff --git a/requirements-dev.txt b/requirements-dev.txt index a3e4d12a3..40b335131 100644 --- a/requirements-dev.txt +++ b/requirements-dev.txt @@ -4,14 +4,14 @@ atomicwrites==1.3.0 cfgv==1.6.0 coverage==4.5.3 coveralls==1.7.0 -django-stubs==0.12.1 +django-stubs==1.2.0 docopt==0.6.2 execnet==1.6.0 identify==1.4.1 importlib-metadata==0.9 importlib-resources==1.0.2 more-itertools==7.0.0 -mypy==0.701 +mypy==0.730 nodeenv==1.3.3 pathlib2==2.3.3 pluggy==0.9.0