doc: Add docstrings throughout.

This commit is contained in:
Brett Smith 2020-11-19 12:00:14 -05:00
parent 0bd6353105
commit e51be066d0
3 changed files with 160 additions and 4 deletions

View file

@ -53,13 +53,28 @@ class PayPalSite(enum.Enum):
class PayPalFields(enum.Flag): class PayPalFields(enum.Flag):
"""Base class for PayPal fields specifiers
Multiple PayPal APIs accept a ``fields`` parameter to let the user specify
what details to return. This class lets code enumerate acceptable values
and combine them programmatically. The ``param_value`` method then helps
``PayPalAPIClient`` format the result when needed.
"""
@classmethod @classmethod
def choices(cls) -> Iterator[str]: def choices(cls) -> Iterator[str]:
"""Iterate the names of all field values"""
for flag in cls: for flag in cls:
yield flag.name.lower() yield flag.name.lower()
@classmethod @classmethod
def combine(cls: Type[FieldsType], fields: Optional[Sequence[FieldsType]]=None) -> FieldsType: def combine(cls: Type[FieldsType], fields: Optional[Sequence[FieldsType]]=None) -> FieldsType:
"""Combine multiple field objects into one
This method just returns the result of ORing all of the fields in the
sequence together. If no argument is given or the sequence is empty,
return the combination of all fields.
"""
if fields: if fields:
fields_iter = iter(fields) fields_iter = iter(fields)
else: else:
@ -71,18 +86,25 @@ class PayPalFields(enum.Flag):
@classmethod @classmethod
def from_arg(cls: Type[FieldsType], arg: str) -> FieldsType: def from_arg(cls: Type[FieldsType], arg: str) -> FieldsType:
"""Return a field object from an argument name string"""
try: try:
return cls[arg.upper()] return cls[arg.upper()]
except KeyError: except KeyError:
raise ValueError(f"unknown {cls.__name__} {arg!r}") from None raise ValueError(f"unknown {cls.__name__} {arg!r}") from None
def is_base_field(self) -> bool: def is_base_field(self) -> bool:
"""Return true if this is a single field value, not a combination"""
return not math.log2(self.value) % 1 return not math.log2(self.value) % 1
def _base_value(self) -> str: def _base_value(self) -> str:
return self.name.lower() return self.name.lower()
def param_value(self) -> str: def param_value(self) -> str:
"""Return these fields formatted as a query string
The result is in the format PayPal's API expects for ``fields``
parameters.
"""
return ','.join( return ','.join(
flag._base_value() flag._base_value()
for flag in type(self) for flag in type(self)
@ -111,6 +133,17 @@ class TransactionFields(PayPalFields):
class PayPalSession(requests_oauthlib.OAuth2Session): class PayPalSession(requests_oauthlib.OAuth2Session):
"""Low-level HTTP session for the PayPal API
This is a subclass of requests_oauthlib.OAuth2Session that implements
PayPal's recommended authorization strategy: if an API request returns
HTTP Unauthorized, get an OAuth token and retry. This gracefully handles
refreshing expired tokens.
This class only handles the mechanics of handling an HTTP connection.
It doesn't know anything about the higher-level REST API. That's the job
of ``PayPalAPIClient``.
"""
TOKEN_PATH = '/v1/oauth2/token' TOKEN_PATH = '/v1/oauth2/token'
def __init__(self, client: oauth2.Client, client_secret: str) -> None: def __init__(self, client: oauth2.Client, client_secret: str) -> None:
@ -130,12 +163,33 @@ class PayPalSession(requests_oauthlib.OAuth2Session):
class PayPalAPIClient: class PayPalAPIClient:
"""Primary access point for the PayPal API
This is the primary class of the library. Most users will instantiate a
``PayPalAPIClient`` using one of the constructor classmethods, then call
methods to make API calls.
"""
def __init__( def __init__(
self, self,
session: requests.Session, session: requests.Session,
root_url: Union[str, PayPalSite]=PayPalSite.SANDBOX, root_url: Union[str, PayPalSite]=PayPalSite.SANDBOX,
logger: Optional[logging.Logger]=None, logger: Optional[logging.Logger]=None,
) -> None: ) -> None:
"""Low-level constructor
``PayPalAPIClient`` expects its underlying ``Session`` object to know
how to authorize itself to PayPal. Usually that means using an instance
of ``PayPalSession``. You can implement and provide your own subclass
of ``requests.Session`` to do this if you prefer.
``root_url`` is either a PayPalSite value, or a string with a full URL
to a PayPal API endpoint.
``logger`` is a ``logging.Logger`` object where all log messages (like
API errors) will be sent. If none is provided, this instance will get
its own, with a name based on the hostname in ``root_url``.
"""
self._session = session self._session = session
if isinstance(root_url, str): if isinstance(root_url, str):
self._root_url = root_url self._root_url = root_url
@ -156,6 +210,15 @@ class PayPalAPIClient:
root_url: Union[str, PayPalSite]=PayPalSite.SANDBOX, root_url: Union[str, PayPalSite]=PayPalSite.SANDBOX,
logger: Optional[logging.Logger]=None, logger: Optional[logging.Logger]=None,
) -> 'PayPalAPIClient': ) -> 'PayPalAPIClient':
"""High-level constructor from individual string arguments
Given ``client_id`` and ``client_secret`` strings, this method
constructs a ``PayPalSesssion`` from them, and then returns a
``PayPalAPIClient`` backed by it.
``root_url`` and ``logger`` arguments are passed directly to
``PayPalAPIClient.__init__``.
"""
client = oauth2.BackendApplicationClient(client_id=client_id) client = oauth2.BackendApplicationClient(client_id=client_id)
session = PayPalSession(client, client_secret) session = PayPalSession(client, client_secret)
return cls(session, root_url, logger) return cls(session, root_url, logger)
@ -167,6 +230,17 @@ class PayPalAPIClient:
default_url: Union[str, PayPalSite]=PayPalSite.SANDBOX, default_url: Union[str, PayPalSite]=PayPalSite.SANDBOX,
logger: Optional[logging.Logger]=None, logger: Optional[logging.Logger]=None,
) -> 'PayPalAPIClient': ) -> 'PayPalAPIClient':
"""High-level constructor from a configuration mapping
Given a mapping of strings (e.g., a configparser section object),
gets the arguments necessary to call ``from_client_secret``, and calls
it.
If the mapping has a ``site`` key, the value will be used as the
``root_url``. Otherwise, ``root_url`` has the value of ``default_url``.
``logger`` is passed directly to ``PayPalAPIClient.__init__``.
"""
try: try:
client_id = config['client_id'] client_id = config['client_id']
client_secret = config['client_secret'] client_secret = config['client_secret']
@ -217,6 +291,7 @@ class PayPalAPIClient:
subscription_id: str, subscription_id: str,
fields: SubscriptionFields=SubscriptionFields.ALL, fields: SubscriptionFields=SubscriptionFields.ALL,
) -> APIResponse: ) -> APIResponse:
"""Fetch and return a subscription by its id"""
return self._get_json(f'/v1/billing/subscriptions/{subscription_id}', { return self._get_json(f'/v1/billing/subscriptions/{subscription_id}', {
'fields': fields.param_value(), 'fields': fields.param_value(),
}) })
@ -228,6 +303,21 @@ class PayPalAPIClient:
start_date: Optional[datetime.datetime]=None, start_date: Optional[datetime.datetime]=None,
fields: TransactionFields=TransactionFields.TRANSACTION, fields: TransactionFields=TransactionFields.TRANSACTION,
) -> APIResponse: ) -> APIResponse:
"""Find and return a transaction by its id
The PayPal API does not provide a way to look up transactions solely by
id. This is a convenience method that wraps the search method to search
different windows of time until it finds the desired transaction.
``start_date`` and ``end_date`` specify the full window of time to
search. This method starts by searching 30 days before ``end_date``,
then the previous 30 days, and so on until it reaches ``start_date``.
The default ``end_date`` is now, and the default ``start_date`` is
three years ago (the API only supports searches this far back).
``fields`` is a TransactionFields object that flags the information to
include in the returned Transaction.
"""
now = datetime.datetime.now(datetime.timezone.utc) now = datetime.datetime.now(datetime.timezone.utc)
if end_date is None: if end_date is None:
end_date = now end_date = now
@ -256,6 +346,12 @@ class PayPalAPIClient:
end_date: datetime.datetime, end_date: datetime.datetime,
fields: TransactionFields=TransactionFields.TRANSACTION, fields: TransactionFields=TransactionFields.TRANSACTION,
) -> Iterator[Transaction]: ) -> Iterator[Transaction]:
"""Iterate transactions over a date range
``start_date`` and ``end_date`` represent the range to query.
``fields`` is a TransactionsFields object that flags the information to
include in returned Transactions.
"""
for page in self._iter_pages('/v1/reporting/transactions', { for page in self._iter_pages('/v1/reporting/transactions', {
'fields': fields.param_value(), 'fields': fields.param_value(),
'start_date': start_date.isoformat(timespec='seconds'), 'start_date': start_date.isoformat(timespec='seconds'),

View file

@ -15,4 +15,11 @@
# along with this program. If not, see <https://www.gnu.org/licenses/>. # along with this program. If not, see <https://www.gnu.org/licenses/>.
class MissingFieldError(KeyError): class MissingFieldError(KeyError):
"""Error raised when code tries to access an unloaded field
This error is raised by PayPal object classes when the caller tries to
access a field that was not loaded in the original API call. For
example, trying to get a payer's name or email address from a Transaction
when the ``fields`` argument did not include ``TransactionFields.PAYER``.
"""
pass pass

View file

@ -79,6 +79,13 @@ class TransactionStatus(enum.Enum):
class Transaction(APIResponse): class Transaction(APIResponse):
"""PayPal Transaction wrapper
The public methods of a Transaction know how to traverse PayPal's JSON
response and turn the data into native Python objects. If Transaction is
missing a method you need, you can also use standarding Mapping methods to
access to raw response.
"""
__slots__ = ['_response'] __slots__ = ['_response']
def __init__(self, response: APIResponse) -> None: def __init__(self, response: APIResponse) -> None:
@ -113,11 +120,28 @@ class Transaction(APIResponse):
return retval return retval
def _wrap_response( # type:ignore[misc] def _wrap_response( # type:ignore[misc]
name: str,
func: Callable[[Any], T], func: Callable[[Any], T],
*keys: str, *keys: str,
doc: Optional[str]=None,
key_doc: Optional[str]=None,
return_doc: Optional[str]=None,
) -> Callable[['Transaction'], T]: ) -> Callable[['Transaction'], T]:
if doc is None:
if key_doc is None:
key_doc = f"``{keys[-1]}``"
if return_doc is None:
if name.endswith('amount'):
return_doc = 'Amount'
elif name.endswith('date'):
return_doc = 'datetime'
else:
return_doc = func.__name__
doc = f"Return the transaction's {key_doc} as a ``{return_doc}``"
def response_wrapper(self: 'Transaction') -> T: def response_wrapper(self: 'Transaction') -> T:
return func(self._get_from_response(*keys)) return func(self._get_from_response(*keys))
response_wrapper.__name__ = name
response_wrapper.__doc__ = doc
return response_wrapper return response_wrapper
def _fee_amount(txn_info: APIResponse) -> Optional[Amount]: # type:ignore[misc] def _fee_amount(txn_info: APIResponse) -> Optional[Amount]: # type:ignore[misc]
@ -129,31 +153,60 @@ class Transaction(APIResponse):
return Amount.from_api(raw_fee) return Amount.from_api(raw_fee)
amount = _wrap_response( amount = _wrap_response(
'amount',
Amount.from_api, Amount.from_api,
'transaction_info', 'transaction_info',
'transaction_amount', 'transaction_amount',
) )
fee_amount = _wrap_response(_fee_amount, 'transaction_info') fee_amount = _wrap_response(
'fee_amount',
_fee_amount,
'transaction_info',
key_doc='``fee_amount``',
)
initiation_date = _wrap_response( initiation_date = _wrap_response(
'initiation_date',
parse_datetime, parse_datetime,
'transaction_info', 'transaction_info',
'transaction_initiation_date', 'transaction_initiation_date',
) )
payer_email = _wrap_response(str, 'payer_info', 'email_address') payer_email = _wrap_response(
payer_fullname = _wrap_response(str, 'payer_info', 'payer_name', 'alternate_full_name') 'payer_email',
str,
'payer_info',
'email_address',
key_doc="payer's email address",
)
payer_fullname = _wrap_response(
'payer_fullname',
str,
'payer_info',
'payer_name',
'alternate_full_name',
key_doc="payer's full name",
)
status = _wrap_response( status = _wrap_response(
'status',
TransactionStatus.__getitem__, TransactionStatus.__getitem__,
'transaction_info', 'transaction_info',
'transaction_status', 'transaction_status',
return_doc=TransactionStatus.__name__,
)
transaction_id = _wrap_response(
'transaction_id',
str,
'transaction_info',
'transaction_id',
) )
transaction_id = _wrap_response(str, 'transaction_info', 'transaction_id')
updated_date = _wrap_response( updated_date = _wrap_response(
'updated_date',
parse_datetime, parse_datetime,
'transaction_info', 'transaction_info',
'transaction_updated_date', 'transaction_updated_date',
) )
def cart_items(self) -> Iterator[CartItem]: def cart_items(self) -> Iterator[CartItem]:
"""Iterate a ``CartItem`` object for each item in the transaction's cart"""
cart_info = self._get_from_response('cart_info') cart_info = self._get_from_response('cart_info')
try: try:
item_seq = cart_info['item_details'] item_seq = cart_info['item_details']