FAQ | This is a LIVE service | Changelog

Commit 0c15e5d5 authored by Robin Goodall's avatar Robin Goodall 💬
Browse files

Cleanup from review and py3.9 typing

parent 757f6491
from typing import Dict, Generator, Iterable, List, Tuple
from collections.abc import Generator, Iterable
import os
import sys
import logging
......@@ -8,7 +8,7 @@ import ibisclient
from .types import (GroupChange, GroupChanges, GroupName, GroupSet,
InstIdSet, LookupInstId,
StudentIdSet, StudentMapping, StudentsByGroup)
StudentId, StudentIdSet, StudentMapping, StudentsByGroup)
LOG = logging.getLogger(os.path.basename(sys.argv[0]))
......@@ -72,7 +72,7 @@ def group_name(instid: LookupInstId, career: str) -> GroupName:
def compare_with_lookup_groups(
ibis_group_methods: ibisclient.GroupMethods,
students_by_group: StudentsByGroup,
student_names_by_id: StudentMapping) -> Tuple[GroupSet, GroupChanges]:
student_names_by_id: StudentMapping) -> tuple[GroupSet, GroupChanges]:
Check each lookup group exists and compare its membership to supplied set.
Provide set of missing Lookup groups, and necessary changes as a dict mapping
......@@ -94,7 +94,7 @@ def compare_with_lookup_groups(
LOG.info('Group "%s" should have %s student(s):', group, len(students))
# Get lookup direct membership
members: List[ibisclient.IbisPerson] = ibis_group_methods.getDirectMembers(
members: list[ibisclient.IbisPerson] = ibis_group_methods.getDirectMembers(
group, 'all_identifiers')
# Get set of USNs from membership
......@@ -123,7 +123,7 @@ def compare_with_lookup_groups(
for id in person.identifiers if id.scheme == 'usn' and id.value in to_remove
if to_add | to_remove:
if to_add or to_remove:
group_changes[group] = GroupChange(add=to_add, remove=to_remove)
LOG.info('%s group(s) need creating', len(missing_groups))
......@@ -131,17 +131,17 @@ def compare_with_lookup_groups(
return (missing_groups, group_changes)
def chunker(n: int, it: Iterable) -> Generator[Tuple[int, StudentIdSet], None, None]:
def chunker(n: int, ids: Iterable[StudentId]) -> Generator[tuple[int, StudentIdSet], None, None]:
Yield chunks of size `n` from `it`. The last chunk may be smaller.
If `n` is not greater than the size of `it`, just yields the one chunk
Yield chunks of size `n` from `ids`. The last chunk may be smaller.
If `n` is not greater than the size of `ids`, just yields the one chunk
if n <= 0:
yield 0, list(it)
yield 0, set(ids)
args = [iter(it)] * n
chunks = itertools.zip_longest(*args, fillvalue=None)
iterators = [iter(ids)] * n
chunks = itertools.zip_longest(*iterators, fillvalue=None)
# unfortunately we need to drop the Nones on the last element
for i, chunk in enumerate(chunks):
yield i, set(filter(None, chunk))
......@@ -158,7 +158,7 @@ def check_usns_in_lookup(
matched_usns: StudentIdSet = set()
LOG.info('Checking USNs exist in Lookup:')
for idx, usns in chunker(CHUNK_SIZE, all_usns_to_add):
people: List[ibisclient.IbisPerson] = ibis_person_methods.listPeople(
people: list[ibisclient.IbisPerson] = ibis_person_methods.listPeople(
[f'usn/{usn}' for usn in usns],
......@@ -186,14 +186,13 @@ def filter_usns_from_changes(
missing_in_group_add = {
usn for usn in changes['add'] if usn in missing_usns
if missing_in_group_add:
LOG.info('Ignoring in group %s:', group)
for usn in missing_in_group_add:
LOG.info('- usn/%s (%s)', usn, student_names_by_id.get(usn, 'Unknown'))
# remove ignored USNs from set to add
changes['add'] = {
usn for usn in changes['add'] if usn not in missing_in_group_add
if not missing_in_group_add:
LOG.info('Ignoring in group %s:', group)
for usn in missing_in_group_add:
LOG.info('- usn/%s (%s)', usn, student_names_by_id.get(usn, 'Unknown'))
# remove ignored USNs from set to add
changes['add'] -= missing_in_group_add
def update_lookup_groups(
......@@ -237,7 +236,7 @@ def create_lookup_groups(
# Get current authenticated lookup account
managed_by = ibis_inst_methods.conn.username
# Cache institution details so we can name, title and describe groups appropriately
institutions: Dict[LookupInstId, ibisclient.IbisInstitution] = {}
institutions: dict[LookupInstId, ibisclient.IbisInstitution] = {}
# Compile a set of institutions we couldn't find
missing_insts: InstIdSet = set()
from typing import Generator, List, Optional, Tuple
from typing import Optional
from collections.abc import Generator
import logging
import os
import sys
......@@ -24,7 +25,7 @@ LOG = logging.getLogger(os.path.basename(sys.argv[0]))
def get_students_by_group(
session: requests_oauthlib.OAuth2Session,
inst_map: InstMapping) -> Tuple[StudentsByGroup, StudentMapping]:
inst_map: InstMapping) -> tuple[StudentsByGroup, StudentMapping]:
Create a map from Lookup group name to sets of students within that institution with status
matching career. Group names are formed from the Lookup instids and student identifiers
......@@ -135,15 +136,15 @@ class Student(pydantic.BaseModel):
Student resource from Student API.
affiliations: List[StudentAffiliation]
affiliations: list[StudentAffiliation]
forenames: str
identifiers: List[StudentIdentifier]
identifiers: list[StudentIdentifier]
namePrefixes: str
surname: str
def fetch_all_students(
session: requests_oauthlib.OAuth2Session) -> Generator[List[Student], None, None]:
session: requests_oauthlib.OAuth2Session) -> Generator[list[Student], None, None]:
Fetch all students from the students API. *Generates* a list of Student resources.
from typing import Dict
from contextlib import contextmanager
from tempfile import NamedTemporaryFile
import os
......@@ -35,7 +34,7 @@ class MockResponse():
class MockSession():
def __init__(self, responses: Dict[str, Dict]):
def __init__(self, responses: dict[str, dict]):
self.responses = responses
self.urls_got = set()
from typing import List, Dict, Optional, Tuple, Union
from typing import Optional, Union
import faker
from identitylib.identifiers import IdentifierSchemes
......@@ -13,7 +13,7 @@ class StudentProvider(faker.providers.BaseProvider):
def student(self, **kwargs) -> Dict[str, Union[str, Dict]]:
def student(self, **kwargs) -> dict[str, Union[str, list, dict]]:
A single fake student, overriding properties as necessary
......@@ -35,13 +35,13 @@ class StudentProvider(faker.providers.BaseProvider):
else (' ' + self.generator.first_name())
def affiliations(self, num_affiliations=1, **kwargs) -> List[Dict[str, str]]:
def affiliations(self, num_affiliations=1, **kwargs) -> list[dict[str, str]]:
return [self.affiliation(**kwargs) for _ in range(num_affiliations)]
def identifiers(self, num_identifiers=1, **kwargs) -> List[Dict[str, str]]:
def identifiers(self, num_identifiers=1, **kwargs) -> list[dict[str, str]]:
return [self.identifier(**kwargs) for _ in range(num_identifiers)]
def affiliation(self, **kwargs) -> Dict[str, str]:
def affiliation(self, **kwargs) -> dict[str, str]:
value, scheme = (kwargs.get('affiliation_id') or self.affiliation_id(**kwargs)).split('@')
start, end = kwargs.get('affiliation_period') or self.affiliation_period(**kwargs)
return {
......@@ -56,14 +56,14 @@ class StudentProvider(faker.providers.BaseProvider):
# Provides example affiliation (in test mapping) with expected scheme
return f'{SR_INST_VALUE}@{IdentifierSchemes.STUDENT_INSTITUTION}'
def affiliation_period(self, **kwargs) -> Tuple[Optional[str], Optional[str]]:
def affiliation_period(self, **kwargs) -> tuple[Optional[str], Optional[str]]:
# Unspecified start and end dates
return (None, None)
def affiliation_status(self, **kwargs) -> str:
return self.random_element(ACADEMIC_CAREER_MAPPING.keys())
def identifier(self, **kwargs) -> Dict[str, str]:
def identifier(self, **kwargs) -> dict[str, str]:
value, scheme = (kwargs.get('identifier_id') or self.identifier_id(**kwargs)).split('@')
return {
'value': value,
import logging
from typing import Dict, List
from unittest.mock import MagicMock, patch
import pytest
......@@ -142,7 +141,7 @@ def mock_lookup_group_members(mock_students_by_group, mock_student_names_by_id):
# Mock ibisclient.GroupMethods
class MockGroupMethods:
def __init__(self, groups: List[Dict[str, ibisclient.IbisPerson]] = list()) -> None:
def __init__(self, groups: list[dict[str, ibisclient.IbisPerson]] = []) -> None:
self.groups = groups
self.updates = {}
......@@ -298,7 +297,7 @@ def test_update_lookup_groups(caplog):
# Mock ibisclient.InstitutionMethods
class MockInstitutionMethods:
def __init__(self, insts: List[Dict[str, ibisclient.IbisInstitution]] = list()) -> None:
def __init__(self, insts: list[dict[str, ibisclient.IbisInstitution]] = []) -> None:
self.insts = insts
self.group_creations = {}
self.conn = MagicMock()
......@@ -431,11 +430,11 @@ def mock_lookup_people(mock_student_names_by_id):
# Mock ibisclient.PersonMethods
class MockPersonMethods:
def __init__(self, people: Dict[str, ibisclient.IbisPerson] = {}) -> None:
def __init__(self, people: dict[str, ibisclient.IbisPerson] = {}) -> None:
self.people = people
self.listCount = 0
def listPeople(self, ids: List[str], fetch: str):
def listPeople(self, ids: list[str], fetch: str):
self.listCount += 1
assert fetch == 'all_identifiers'
assert all([id[:4] == 'usn/' for id in ids])
from typing import List, Dict
import logging
from unittest import TestCase
from urllib.parse import urljoin
......@@ -113,7 +112,7 @@ class StudentAPITest(TestCase):
Student.parse_obj(s) for s in fake_students
def _students_to_responses(self, students: List[Dict]) -> Dict[str, Dict]:
def _students_to_responses(self, students: list[dict]) -> dict[str, dict]:
Take a list of students and return a single page response for the initial
page request
# Convenient type aliases and definitions
from typing import Dict, Set, TypedDict
from typing import TypedDict
# Institution Ids in both Student API and Lookup are strings and we use a mapping between them
StudentAPIInstId = str
LookupInstId = str
InstMapping = Dict[StudentAPIInstId, LookupInstId]
InstIdSet = Set[LookupInstId]
InstMapping = dict[StudentAPIInstId, LookupInstId]
InstIdSet = set[LookupInstId]
# Student USNs and names are strings, and for logging we keep a mapping between them
StudentId = str
StudentName = str
StudentMapping = Dict[StudentId, StudentName]
StudentIdSet = Set[StudentId]
StudentMapping = dict[StudentId, StudentName]
StudentIdSet = set[StudentId]
# Group names as strings, and we use a mapping of group names to sets of student ids
GroupName = str
StudentsByGroup = Dict[GroupName, StudentIdSet]
GroupSet = Set[GroupName]
StudentsByGroup = dict[GroupName, StudentIdSet]
GroupSet = set[GroupName]
# A group change consists of a set of student ids to add and a set to remove
......@@ -28,4 +28,4 @@ class GroupChange(TypedDict):
# A collection of group changes is a mapping of group name to changes
GroupChanges = Dict[GroupName, GroupChange]
GroupChanges = dict[GroupName, GroupChange]
Supports Markdown
0% or .
You are about to add 0 people to the discussion. Proceed with caution.
Finish editing this message first!
Please register or to comment