Skip to content

Commit

Permalink
Changes requested by linters
Browse files Browse the repository at this point in the history
* Updates to type hinting, code structure, datetime objects, unit tests, fixtures, and docstrings as requested by linters
  • Loading branch information
ehanson8 committed Dec 5, 2023
1 parent c8f47c2 commit 610e794
Show file tree
Hide file tree
Showing 11 changed files with 103 additions and 111 deletions.
20 changes: 8 additions & 12 deletions ccslips/alma.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,6 @@
import logging
import time
from typing import Generator, Optional
from collections.abc import Generator
from urllib.parse import urljoin

import requests
Expand Down Expand Up @@ -39,7 +39,7 @@ def get_paged(
self,
endpoint: str,
record_type: str,
params: Optional[dict] = None,
params: dict | None = None,
limit: int = 100,
_offset: int = 0,
_records_retrieved: int = 0,
Expand Down Expand Up @@ -75,8 +75,7 @@ def get_paged(
total_record_count = int(response.json()["total_record_count"])
records = response.json().get(record_type, [])
records_retrieved = _records_retrieved + len(records)
for record in records:
yield record
yield from records
if records_retrieved < total_record_count:
yield from self.get_paged(
endpoint,
Expand All @@ -88,10 +87,9 @@ def get_paged(
)

def get_brief_po_lines(
self, acquisition_method: Optional[str] = None
self, acquisition_method: str | None = None
) -> Generator[dict, None, None]:
"""
Get brief PO line records, optionally filtered by acquisition_method.
"""Get brief PO line records, optionally filtered by acquisition_method.
The PO line records retrieved from this endpoint do not contain all of the PO
line data and users may wish to retrieve the full PO line records with the
Expand All @@ -118,15 +116,13 @@ def get_full_po_line(self, po_line_id: str) -> dict:

def get_full_po_lines(
self,
acquisition_method: Optional[str] = None,
date: Optional[str] = None,
acquisition_method: str | None = None,
date: str | None = None,
) -> Generator[dict, None, None]:
"""Get full PO line records, optionally filtered by acquisition_method/date."""
for line in self.get_brief_po_lines(acquisition_method):
number = line["number"]
if date is None:
yield self.get_full_po_line(number)
elif line.get("created_date") == f"{date}Z":
if line.get("created_date") == f"{date}Z" or not date:
yield self.get_full_po_line(number)

def get_fund_by_code(self, fund_code: str) -> dict:
Expand Down
13 changes: 7 additions & 6 deletions ccslips/cli.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,7 @@
import datetime
import logging
import os
from datetime import datetime, timedelta
from time import perf_counter
from typing import Optional

import click

Expand Down Expand Up @@ -52,8 +51,8 @@ def main(
ctx: click.Context,
source_email: str,
recipient_email: list[str],
date: Optional[str],
log_level: Optional[str],
date: str | None,
log_level: str | None,
) -> None:
start_time = perf_counter()
log_level = log_level or "INFO"
Expand All @@ -63,7 +62,9 @@ def main(
logger.debug("Command called with options: %s", ctx.params)

logger.info("Starting credit card slips process")
date = date or (datetime.today() - timedelta(days=2)).strftime("%Y-%m-%d")
date = date or (
datetime.datetime.now(tz=datetime.UTC) - datetime.timedelta(days=2)
).strftime("%Y-%m-%d")
credit_card_slips_data = process_po_lines(date)
email_content = generate_credit_card_slips_html(credit_card_slips_data)
email = Email()
Expand All @@ -90,5 +91,5 @@ def main(
date,
recipient_email,
response["MessageId"],
str(timedelta(seconds=elapsed_time)),
str(datetime.timedelta(seconds=elapsed_time)),
)
5 changes: 3 additions & 2 deletions ccslips/config.py
Original file line number Diff line number Diff line change
Expand Up @@ -6,9 +6,10 @@

def configure_logger(logger: logging.Logger, log_level_string: str) -> str:
if log_level_string.upper() not in logging.getLevelNamesMapping():
raise ValueError(f"'{log_level_string}' is not a valid Python logging level")
message = f"'{log_level_string}' is not a valid Python logging level"
raise ValueError(message)
log_level = logging.getLevelName(log_level_string.upper())
if log_level < 20:
if log_level < 20: # noqa: PLR2004
logging.basicConfig(
format="%(asctime)s %(levelname)s %(name)s.%(funcName)s() line %(lineno)d: "
"%(message)s"
Expand Down
16 changes: 7 additions & 9 deletions ccslips/email.py
Original file line number Diff line number Diff line change
@@ -1,6 +1,5 @@
from email.message import EmailMessage
from email.policy import EmailPolicy, default
from typing import List, Optional

import boto3

Expand All @@ -12,16 +11,16 @@ def __init__(self, policy: EmailPolicy = default) -> None:
"""Initialize Email instance."""
super().__init__(policy)

def populate( # noqa pylint R0913 too many arguments
def populate(
self,
from_address: str,
to_addresses: str,
subject: str,
attachments: Optional[List[dict]] = None,
body: Optional[str] = None,
bcc: Optional[str] = None,
cc: Optional[str] = None, # noqa pylint C0103 not snake_case
reply_to: Optional[str] = None,
attachments: list[dict] | None = None,
body: str | None = None,
bcc: str | None = None,
cc: str | None = None,
reply_to: str | None = None,
) -> None:
"""Populate Email message with addresses and subject.
Expand Down Expand Up @@ -70,11 +69,10 @@ def send(self) -> dict[str, str]:
destinations.extend(self["Cc"].split(","))
if self["Bcc"]:
destinations.extend(self["Bcc"].split(","))
response = ses.send_raw_email(
return ses.send_raw_email(
Source=self["From"],
Destinations=destinations,
RawMessage={
"Data": self.as_bytes(),
},
)
return response
42 changes: 21 additions & 21 deletions ccslips/polines.py
Original file line number Diff line number Diff line change
@@ -1,8 +1,8 @@
import xml.etree.ElementTree as ET # nosec B405
from collections.abc import Generator, Iterator
from copy import deepcopy
from datetime import datetime
from decimal import Decimal
from typing import Generator, Iterator, Optional
from xml.etree import ElementTree

from ccslips.alma import AlmaClient

Expand All @@ -20,9 +20,11 @@ def extract_credit_card_slip_data(client: AlmaClient, po_line_record: dict) -> d
The keys of the returned dict map to the appropriate element classes in the XML
template used to generate a formatted slip.
"""
created_date = datetime.strptime(
po_line_record["created_date"], "%Y-%m-%dZ"
).strftime("%y%m%d")
created_date = (
datetime.strptime(po_line_record["created_date"], "%Y-%m-%dZ")
.astimezone()
.strftime("%y%m%d")
)
fund_distribution = po_line_record.get("fund_distribution", [])
price = Decimal(po_line_record.get("price", {}).get("sum", "0.00"))
title = po_line_record.get("resource_metadata", {}).get("title", "Unknown title")
Expand All @@ -47,15 +49,15 @@ def extract_credit_card_slip_data(client: AlmaClient, po_line_record: dict) -> d
return po_line_data


def get_cardholder_from_notes(notes: Optional[list[dict]]) -> str:
def get_cardholder_from_notes(notes: list[dict] | None) -> str:
"""Get first note that begins with 'CC-' from a PO line record notes field."""
if notes:
for note in [n for n in notes if n.get("note_text", "").startswith("CC-")]:
return note["note_text"][3:]
return "No cardholder note found"


def get_quantity_from_locations(locations: Optional[list[dict]]) -> str:
def get_quantity_from_locations(locations: list[dict] | None) -> str:
"""Get the total quantity of items associated with PO line locations.
This function adds the quantities from each location in the PO line. This is an
Expand Down Expand Up @@ -87,53 +89,51 @@ def get_total_price_from_fund_distribution(
)


def get_account_data(
client: AlmaClient, fund_distribution: list[dict]
) -> dict[str, str]:
def get_account_data(client: AlmaClient, fund_distribution: list[dict]) -> dict[str, str]:
"""Get account information needed for a credit card slip.
If the fund_distribution is empty, returns a single account with default text.
Otherwise returns up to two accounts with their associated account numbers.
"""
result = {"account_1": "No fund code found"}
for count, fund in enumerate(fund_distribution, start=1):
if count == 3:
if count == 3: # noqa: PLR2004
break
if account_number := get_account_number_from_fund(client, fund):
result[f"account_{count}"] = account_number
return result


def get_account_number_from_fund(client: AlmaClient, fund: dict) -> Optional[str]:
def get_account_number_from_fund(client: AlmaClient, fund: dict) -> str | None:
"""Get account number for a given fund.
Returns None if fund has no fund code, fund record cannot be retrieved via fund
code, or fund record does not contain an external_id field value.
"""
account = None
if fund_code := fund.get("fund_code", {}).get("value"):
if fund_code := fund.get("fund_code", {}).get("value"): # noqa: SIM102
if fund_records := client.get_fund_by_code(fund_code).get("fund"):
account = fund_records[0].get("external_id")
return account


def generate_credit_card_slips_html(po_line_data: Iterator[dict]) -> str:
"""Create credit card slips HTML from a set of credit card slip data."""
template_tree = ET.parse("config/credit_card_slip_template.xml") # nosec B314
template_tree = ElementTree.parse( # noqa: S314
"config/credit_card_slip_template.xml"
)
xml_template = template_tree.getroot()
output = ET.fromstring("<html></html>") # nosec B314
output = ElementTree.fromstring("<html></html>") # noqa: S314
for line in po_line_data:
output.append(
populate_credit_card_slip_xml_fields(deepcopy(xml_template), line)
)
output.append(populate_credit_card_slip_xml_fields(deepcopy(xml_template), line))
if len(output) == 0:
return "<html><p>No credit card orders on this date</p></html>"
return ET.tostring(output, encoding="unicode", method="xml")
return ElementTree.tostring(output, encoding="unicode", method="xml")


def populate_credit_card_slip_xml_fields(
credit_card_slip_xml_template: ET.Element, credit_card_slip_data: dict
) -> ET.Element:
credit_card_slip_xml_template: ElementTree.Element, credit_card_slip_data: dict
) -> ElementTree.Element:
"""Populate credit card slip XML template with data extracted from a PO line.
The credit_card_slip_data keys must correspond to their associated element classes
Expand Down
38 changes: 16 additions & 22 deletions tests/conftest.py
Original file line number Diff line number Diff line change
@@ -1,5 +1,4 @@
import json
import os

import boto3
import pytest
Expand All @@ -12,29 +11,24 @@

# Env fixtures
@pytest.fixture(autouse=True)
def test_env():
os.environ = {
"ALMA_API_URL": "https://example.com",
"ALMA_API_READ_KEY": "just-for-testing",
"ALMA_API_TIMEOUT": "10",
"SES_SEND_FROM_EMAIL": "from@example.com",
"SES_RECIPIENT_EMAIL": "recipient1@example.com recipient2@example.com",
"SENTRY_DSN": "None",
"WORKSPACE": "test",
}
yield


@pytest.fixture(autouse=True)
def aws_credentials():
os.environ["AWS_ACCESS_KEY_ID"] = "testing"
os.environ["AWS_SECRET_ACCESS_KEY"] = "testing"
os.environ["AWS_SECURITY_TOKEN"] = "testing"
os.environ["AWS_SESSION_TOKEN"] = "testing"
def _test_environment(monkeypatch):
monkeypatch.setenv("ALMA_API_URL", "https://example.com")
monkeypatch.setenv("ALMA_API_READ_KEY", "just-for-testing")
monkeypatch.setenv("ALMA_API_TIMEOUT", "10")
monkeypatch.setenv("SES_SEND_FROM_EMAIL", "from@example.com")
monkeypatch.setenv(
"SES_RECIPIENT_EMAIL", "recipient1@example.com recipient2@example.com"
)
monkeypatch.setenv("SENTRY_DSN", "None")
monkeypatch.setenv("WORKSPACE", "test")
monkeypatch.setenv("AWS_ACCESS_KEY_ID", "testing")
monkeypatch.setenv("AWS_SECRET_ACCESS_KEY", "testing")
monkeypatch.setenv("AWS_SECURITY_TOKEN", "testing")
monkeypatch.setenv("AWS_SESSION_TOKEN", "testing")


# CLI fixture
@pytest.fixture()
@pytest.fixture
def runner():
return CliRunner()

Expand All @@ -53,7 +47,7 @@ def po_line_records_fixture():


# API fixtures
@pytest.fixture()
@pytest.fixture
def alma_client():
return AlmaClient()

Expand Down
8 changes: 4 additions & 4 deletions tests/test_alma.py
Original file line number Diff line number Diff line change
Expand Up @@ -9,7 +9,7 @@ def test_client_initializes_with_expected_values():
"Accept": "application/json",
"Content-Type": "application/json",
}
assert client.timeout == 10
assert client.timeout == 10 # noqa: PLR2004


def test_get_paged(alma_client):
Expand All @@ -18,7 +18,7 @@ def test_get_paged(alma_client):
record_type="fake_records",
limit=10,
)
assert len(list(records)) == 15
assert len(list(records)) == 15 # noqa: PLR2004


def test_get_brief_po_lines_without_acquisition_method(alma_client):
Expand All @@ -29,7 +29,7 @@ def test_get_brief_po_lines_without_acquisition_method(alma_client):

def test_get_brief_po_lines_with_acquisition_method(alma_client):
result = list(alma_client.get_brief_po_lines("PURCHASE_NOLETTER"))
assert len(result) == 3
assert len(result) == 3 # noqa: PLR2004
assert result[0]["number"] == "POL-all-fields"
assert result[1]["number"] == "POL-missing-fields"
assert result[2]["number"] == "POL-wrong-date"
Expand Down Expand Up @@ -64,7 +64,7 @@ def test_get_full_po_lines_with_parameters(alma_client):
acquisition_method="PURCHASE_NOLETTER", date="2023-01-02"
)
)
assert len(result) == 2
assert len(result) == 2 # noqa: PLR2004
assert result[0]["number"] == "POL-all-fields"
assert result[1]["number"] == "POL-missing-fields"

Expand Down
Loading

0 comments on commit 610e794

Please sign in to comment.