<s> import os <EOL> import glob <EOL> import errno <EOL> import logging <EOL> import functools <EOL> import contextlib <EOL> import subprocess <EOL> from celery . utils . log import get_task_logger <EOL> from waterbutler . tasks . app import app , client <EOL> from waterbutler . providers . osfstorage import settings <EOL> from waterbutler . providers . osfstorage . tasks import exceptions <EOL> logger = get_task_logger ( __name__ ) <EOL> logger . setLevel ( logging . INFO ) <EOL> def ensure_path ( path ) : <EOL> try : <EOL> os . makedirs ( path ) <EOL> except OSError as exception : <EOL> if exception . errno != errno . EEXIST : <EOL> raise <EOL> def ensure_paths ( ) : <EOL> paths = [ <EOL> settings . FILE_PATH_PENDING , <EOL> settings . FILE_PATH_COMPLETE , <EOL> ] <EOL> for path in paths : <EOL> ensure_path ( path ) <EOL> def create_parity_files ( file_path , redundancy = <NUM_LIT:5> ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> stat = os . stat ( file_path ) <EOL> if not stat . st_size : <EOL> return [ ] <EOL> except OSError as error : <EOL> raise exceptions . ParchiveError ( '<STR_LIT>' . format ( error . strerror ) ) <EOL> path , name = os . path . split ( file_path ) <EOL> with open ( os . devnull , '<STR_LIT:wb>' ) as DEVNULL : <EOL> args = [ <EOL> '<STR_LIT>' , <EOL> '<STR_LIT:c>' , <EOL> '<STR_LIT>' . format ( redundancy ) , <EOL> os . path . join ( path , '<STR_LIT>' . format ( name ) ) , <EOL> file_path , <EOL> ] <EOL> ret_code = subprocess . call ( args , stdout = DEVNULL , stderr = DEVNULL ) <EOL> if ret_code != <NUM_LIT:0> : <EOL> raise exceptions . ParchiveError ( '<STR_LIT>' . format ( '<STR_LIT:U+0020>' . join ( args ) , ret_code ) ) <EOL> return [ <EOL> os . path . abspath ( fpath ) <EOL> for fpath in <EOL> glob . glob ( os . path . join ( path , '<STR_LIT>' . format ( name ) ) ) <EOL> ] <EOL> def sanitize_request ( request ) : <EOL> """<STR_LIT>""" <EOL> return { <EOL> key : value <EOL> for key , value in vars ( request ) . items ( ) <EOL> if key not in [ '<STR_LIT:args>' , '<STR_LIT>' ] <EOL> } <EOL> def _log_task ( func ) : <EOL> """<STR_LIT>""" <EOL> @ functools . wraps ( func ) <EOL> def wrapped ( self , * args , ** kwargs ) : <EOL> logger . info ( sanitize_request ( self . request ) ) <EOL> return func ( self , * args , ** kwargs ) <EOL> return wrapped <EOL> def _create_task ( * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> def wrapper ( func ) : <EOL> wrapped = _log_task ( func ) <EOL> wrapped = app . task ( bind = True , * args , ** kwargs ) ( wrapped ) <EOL> return wrapped <EOL> return wrapper <EOL> def task ( * args , ** kwargs ) : <EOL> """<STR_LIT>""" <EOL> if len ( args ) == <NUM_LIT:1> and callable ( args [ <NUM_LIT:0> ] ) : <EOL> return _create_task ( ) ( args [ <NUM_LIT:0> ] ) <EOL> return _create_task ( * args , ** kwargs ) <EOL> def get_countdown ( attempt , init_delay , max_delay , backoff ) : <EOL> multiplier = backoff ** attempt <EOL> return min ( init_delay * multiplier , max_delay ) <EOL> def capture_retry_message ( task ) : <EOL> if not client : <EOL> return <EOL> client . captureException ( extra = sanitize_request ( task . request ) ) <EOL> @ contextlib . contextmanager <EOL> def RetryTask ( task , attempts , init_delay , max_delay , backoff , warn_idx , error_types ) : <EOL> try : <EOL> yield <EOL> except error_types as exc_value : <EOL> try_count = task . request . retries <EOL> if warn_idx is not None and try_count >= warn_idx : <EOL> capture_retry_message ( task ) <EOL> countdown = get_countdown ( try_count , init_delay , max_delay , backoff ) <EOL> task . max_retries = attempts <EOL> raise task . retry ( exc = exc_value , countdown = countdown ) <EOL> RetryUpload = functools . partial ( <EOL> RetryTask , <EOL> attempts = settings . UPLOAD_RETRY_ATTEMPTS , <EOL> init_delay = settings . UPLOAD_RETRY_INIT_DELAY , <EOL> max_delay = settings . UPLOAD_RETRY_MAX_DELAY , <EOL> backoff = settings . UPLOAD_RETRY_BACKOFF , <EOL> warn_idx = settings . UPLOAD_RETRY_WARN_IDX , <EOL> error_types = ( Exception , ) , <EOL> ) <EOL> RetryHook = functools . partial ( <EOL> RetryTask , <EOL> attempts = settings . HOOK_RETRY_ATTEMPTS , <EOL> init_delay = settings . HOOK_RETRY_INIT_DELAY , <EOL> max_delay = settings . HOOK_RETRY_MAX_DELAY , <EOL> backoff = settings . HOOK_RETRY_BACKOFF , <EOL> warn_idx = settings . UPLOAD_RETRY_WARN_IDX , <EOL> error_types = ( Exception , ) , <EOL> ) <EOL> RetryParity = functools . partial ( <EOL> RetryTask , <EOL> attempts = settings . PARITY_RETRY_ATTEMPTS , <EOL> init_delay = settings . PARITY_RETRY_INIT_DELAY , <EOL> max_delay = settings . PARITY_RETRY_MAX_DELAY , <EOL> backoff = settings . PARITY_RETRY_BACKOFF , <EOL> warn_idx = settings . PARITY_RETRY_WARN_IDX , <EOL> error_types = ( Exception , ) , <EOL> ) </s>
