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()) |