Generic Python stuff I find myself rewriting frequently.
No framework-specific functionality here. Any framework-specific dealings will be separated out into their own gists.
""" | |
custom exceptions | |
""" | |
class ExtendedException(Exception): | |
"""Base class for custom exceptions.""" | |
def __init__(self, message, **kwargs): | |
self.__dict__.update(kwargs) | |
Exception.__init__(self, message) |
""" | |
http request handlers | |
""" | |
import requests | |
def handle_get_request(r, silently_fail=False): | |
""" | |
Exception handling for GET request. | |
Set silently_fail=True if we don't care about the output. | |
""" | |
try: | |
r.raise_for_status() | |
except requests.exceptions.RequestException: | |
logger.error( | |
"HTTP request failed", | |
status_code=r.status_code, | |
url=r.url, | |
) | |
if silently_fail: | |
return None | |
raise | |
return r.json().get('data') | |
def handle_post_request(r, silently_fail=False): | |
""" | |
Exception handling for POST request. | |
Set silently_fail=True if we don't care about the output. | |
""" | |
try: | |
r.raise_for_status() | |
except requests.exceptions.RequestException: | |
logger.error( | |
"HTTP request failed", | |
status_code=r.status_code, | |
url=r.url, | |
) | |
if not silently_fail: | |
raise | |
# TODO: response data |
""" | |
custom logging functionality | |
""" | |
import logging | |
def setup_logger(name): | |
""" | |
Set up a logger that dumps extra kwargs to message text. | |
To use in other files: `setup_logger(__name__)` | |
""" | |
logger = logging.getLogger(name) | |
logger.setLevel(logging.DEBUG) | |
# Create console handler | |
ch = logging.StreamHandler() | |
if dxd.config.is_development(): | |
ch.setLevel(logging.INFO) | |
elif dxd.config.is_test(): | |
ch.setLevel(logging.DEBUG) | |
else: | |
ch.setLevel(logging.ERROR) | |
# Create file handler that logs level=ERROR | |
eh = logging.FileHandler('error.log') | |
eh.setLevel(logging.ERROR) | |
# Create formatter | |
formatter = ExtraFormatter( | |
'%(asctime)s - %(name)s - %(levelname)s - %(message)s') | |
# Add formatters to handlers | |
ch.setFormatter(formatter) | |
eh.setFormatter(formatter) | |
# Add handlers to logger | |
logger.addHandler(ch) | |
logger.addHandler(eh) | |
return logger | |
class ExtraFormatter(logging.Formatter): | |
"""Logging formatter that dumps extra kwargs into message.""" | |
BLANK_LOG_RECORD = logging.LogRecord(*[None] * 7) | |
def format(self, record: logging.LogRecord) -> str: | |
""" | |
Format a record to dump out extra kwargs. | |
Bit of a hack, since new args cannot be distinguished. We | |
compare this record against a blank dummy record. | |
""" | |
extra = { | |
k: v for k, v in record.__dict__.items() | |
if k not in self.BLANK_LOG_RECORD.__dict__ and k != 'message' | |
} | |
extra_msg = ', '.join(['%s=%s' % (k, v) for k, v in extra.items()]) | |
return super().format(record) + ' ' + extra_msg |
""" | |
unittest extended functionality | |
""" | |
import unittest | |
class TestCase(unittest.TestCase): | |
"""Extended unittest case class to inherit from.""" | |
def assertRaises(self, exc_cls, func, args=None, kwargs=None, msg=None): | |
""" | |
Clobber default assertRaises method so we can return message on | |
failure & allow kwargs. | |
""" | |
if args is None: | |
args = [] | |
if kwargs is None: | |
kwargs = {} | |
if msg is None: | |
msg = 'Expected exception %s' % exc_cls.__name__ | |
try: | |
func(*args, **kwargs) | |
except exc_cls as e: # success | |
return | |
raise AssertionError(msg) |
Generic Python stuff I find myself rewriting frequently.
No framework-specific functionality here. Any framework-specific dealings will be separated out into their own gists.
""" | |
generic utility functions | |
""" | |
import datetime | |
import json | |
import os | |
def get_dotted_attr_from_dict(d, attr_name): | |
""" | |
Given a dot-separated nested dictionary attribute name, get it from a dict. | |
""" | |
attrs = attr_name.split('.') | |
for a in attrs: | |
d = d.get(a) | |
if d is None: | |
return d | |
return d | |
def pluck_dict(d, fields): | |
"""Return dict with only the specified fields included.""" | |
return {attr: d.get(attr) for attr in fields} | |
def get_epoch(): | |
"""Get current epoch.""" | |
return int(datetime.datetime.utcnow().timestamp()) | |
def get_json_from_file(relative_path, runner=None): | |
"""Load & return the contents of a relative JSON file.""" | |
if runner is None: | |
runner = __file__ | |
runner_dir = os.path.dirname(os.path.realpath(__file__)) | |
with open(os.path.join(runner_dir, relative_path), 'r') as f: | |
return json.loads(f.read()) |