<s> """<STR_LIT>""" <EOL> import datetime <EOL> import json <EOL> import logging <EOL> import math <EOL> import time <EOL> import pprint <EOL> import uuid <EOL> from pamqp import body <EOL> from pamqp import header <EOL> from pamqp import specification <EOL> from rabbitpy import base <EOL> from rabbitpy import exceptions <EOL> from rabbitpy import utils <EOL> LOGGER = logging . getLogger ( __name__ ) <EOL> try : <EOL> py26mv = memoryview ( b'<STR_LIT:foo>' ) <EOL> except NameError : <EOL> class memoryview ( object ) : <EOL> pass <EOL> class Properties ( specification . Basic . Properties ) : <EOL> """<STR_LIT>""" <EOL> pass <EOL> class Message ( base . AMQPClass ) : <EOL> """<STR_LIT>""" <EOL> method = None <EOL> name = '<STR_LIT>' <EOL> def __init__ ( self , channel , body_value , properties = None , <EOL> auto_id = False , opinionated = False ) : <EOL> """<STR_LIT>""" <EOL> super ( Message , self ) . __init__ ( channel , '<STR_LIT>' ) <EOL> self . properties = properties or { } <EOL> if isinstance ( body_value , memoryview ) : <EOL> self . body = bytes ( body_value ) <EOL> else : <EOL> self . body = self . _auto_serialize ( body_value ) <EOL> if ( opinionated or auto_id ) and '<STR_LIT>' not in self . properties : <EOL> if auto_id : <EOL> raise DeprecationWarning ( '<STR_LIT>' ) <EOL> self . _add_auto_message_id ( ) <EOL> if opinionated : <EOL> if '<STR_LIT>' not in self . properties : <EOL> self . _add_timestamp ( ) <EOL> if '<STR_LIT>' in self . properties : <EOL> self . properties [ '<STR_LIT>' ] = self . _as_datetime ( self . properties [ '<STR_LIT>' ] ) <EOL> if self . _invalid_properties : <EOL> msg = '<STR_LIT>' % self . _invalid_properties [ <NUM_LIT:0> ] <EOL> raise KeyError ( msg ) <EOL> @ property <EOL> def delivery_tag ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . method . delivery_tag if self . method else None <EOL> @ property <EOL> def redelivered ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . method . redelivered if self . method else None <EOL> @ property <EOL> def routing_key ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . method . routing_key if self . method else None <EOL> @ property <EOL> def exchange ( self ) : <EOL> """<STR_LIT>""" <EOL> return self . method . exchange if self . method else None <EOL> def ack ( self , all_previous = False ) : <EOL> """<STR_LIT>""" <EOL> if not self . method : <EOL> raise exceptions . ActionException ( '<STR_LIT>' <EOL> '<STR_LIT:message>' ) <EOL> basic_ack = specification . Basic . Ack ( self . method . delivery_tag , <EOL> multiple = all_previous ) <EOL> self . channel . write_frame ( basic_ack ) <EOL> def json ( self ) : <EOL> """<STR_LIT>""" <EOL> try : <EOL> return json . loads ( self . body ) <EOL> except TypeError : <EOL> return json . loads ( self . body . decode ( '<STR_LIT:utf-8>' ) ) <EOL> def nack ( self , requeue = False , all_previous = False ) : <EOL> """<STR_LIT>""" <EOL> if not self . method : <EOL> raise exceptions . ActionException ( '<STR_LIT>' <EOL> '<STR_LIT:message>' ) <EOL> basic_nack = specification . Basic . Nack ( self . method . delivery_tag , <EOL> requeue = requeue , <EOL> multiple = all_previous ) <EOL> self . channel . write_frame ( basic_nack ) <EOL> def pprint ( self , properties = False ) : <EOL> """<STR_LIT>""" <EOL> print ( '<STR_LIT>' % self . method . exchange ) <EOL> print ( '<STR_LIT>' % self . method . routing_key ) <EOL> if properties : <EOL> print ( '<STR_LIT>' ) <EOL> pprint . pprint ( self . properties ) <EOL> print ( '<STR_LIT>' ) <EOL> pprint . pprint ( self . body ) <EOL> def publish ( self , exchange , routing_key = '<STR_LIT>' , mandatory = False , <EOL> immediate = False ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( exchange , base . AMQPClass ) : <EOL> exchange = exchange . name <EOL> payload = utils . maybe_utf8_encode ( self . body ) <EOL> frames = [ specification . Basic . Publish ( exchange = exchange , <EOL> routing_key = routing_key or '<STR_LIT>' , <EOL> mandatory = mandatory , <EOL> immediate = immediate ) , <EOL> header . ContentHeader ( body_size = len ( payload ) , <EOL> properties = self . _properties ) ] <EOL> pieces = int ( math . ceil ( len ( payload ) / <EOL> float ( self . channel . maximum_frame_size ) ) ) <EOL> for offset in range ( <NUM_LIT:0> , pieces ) : <EOL> start = self . channel . maximum_frame_size * offset <EOL> end = start + self . channel . maximum_frame_size <EOL> if end > len ( payload ) : <EOL> end = len ( payload ) <EOL> frames . append ( body . ContentBody ( payload [ start : end ] ) ) <EOL> self . channel . write_frames ( frames ) <EOL> if self . channel . publisher_confirms : <EOL> response = self . channel . wait_for_confirmation ( ) <EOL> if isinstance ( response , specification . Basic . Ack ) : <EOL> return True <EOL> elif isinstance ( response , specification . Basic . Nack ) : <EOL> return False <EOL> else : <EOL> raise exceptions . UnexpectedResponseError ( response ) <EOL> def reject ( self , requeue = False ) : <EOL> """<STR_LIT>""" <EOL> if not self . method : <EOL> raise exceptions . ActionException ( '<STR_LIT>' <EOL> '<STR_LIT:message>' ) <EOL> basic_reject = specification . Basic . Reject ( self . method . delivery_tag , <EOL> requeue = requeue ) <EOL> self . channel . write_frame ( basic_reject ) <EOL> def _add_auto_message_id ( self ) : <EOL> """<STR_LIT>""" <EOL> self . properties [ '<STR_LIT>' ] = str ( uuid . uuid4 ( ) ) <EOL> def _add_timestamp ( self ) : <EOL> """<STR_LIT>""" <EOL> self . properties [ '<STR_LIT>' ] = datetime . datetime . now ( ) <EOL> def _as_datetime ( self , value ) : <EOL> """<STR_LIT>""" <EOL> if value is None : <EOL> return None <EOL> if isinstance ( value , datetime . datetime ) : <EOL> return value <EOL> if isinstance ( value , time . struct_time ) : <EOL> return datetime . datetime ( * value [ : <NUM_LIT:6> ] ) <EOL> if utils . is_string ( value ) : <EOL> value = int ( value ) <EOL> if isinstance ( value , float ) or isinstance ( value , int ) : <EOL> return datetime . datetime . fromtimestamp ( value ) <EOL> raise TypeError ( '<STR_LIT>' % <EOL> type ( value ) ) <EOL> def _auto_serialize ( self , body_value ) : <EOL> """<STR_LIT>""" <EOL> if isinstance ( body_value , dict ) or isinstance ( body_value , list ) : <EOL> self . properties [ '<STR_LIT>' ] = '<STR_LIT:application/json>' <EOL> return json . dumps ( body_value , ensure_ascii = False ) <EOL> return body_value <EOL> def _coerce_properties ( self ) : <EOL> """<STR_LIT>""" <EOL> for key , value in self . properties . items ( ) : <EOL> _type = specification . Basic . Properties . type ( key ) <EOL> if self . properties [ key ] is None : <EOL> continue <EOL> if _type == '<STR_LIT>' : <EOL> if not utils . is_string ( value ) : <EOL> LOGGER . warning ( '<STR_LIT>' , key ) <EOL> value = str ( value ) <EOL> self . properties [ key ] = utils . maybe_utf8_encode ( value ) <EOL> elif _type == '<STR_LIT>' and not isinstance ( value , int ) : <EOL> LOGGER . warning ( '<STR_LIT>' , key ) <EOL> try : <EOL> self . properties [ key ] = int ( value ) <EOL> except TypeError as error : <EOL> LOGGER . warning ( '<STR_LIT>' , key , error ) <EOL> elif _type == '<STR_LIT>' and not isinstance ( value , dict ) : <EOL> LOGGER . warning ( '<STR_LIT>' , key ) <EOL> self . properties [ key ] = { } <EOL> if key == '<STR_LIT>' : <EOL> self . properties [ key ] = self . _as_datetime ( value ) <EOL> @ property <EOL> def _invalid_properties ( self ) : <EOL> """<STR_LIT>""" <EOL> return [ key for key in self . properties <EOL> if key not in specification . Basic . Properties . attributes ( ) ] <EOL> @ property <EOL> def _properties ( self ) : <EOL> """<STR_LIT>""" <EOL> self . _prune_invalid_properties ( ) <EOL> self . _coerce_properties ( ) <EOL> return specification . Basic . Properties ( ** self . properties ) <EOL> def _prune_invalid_properties ( self ) : <EOL> """<STR_LIT>""" <EOL> for key in self . _invalid_properties : <EOL> LOGGER . warning ( '<STR_LIT>' , key ) <EOL> del self . properties [ key ] </s>
