entity.py 32 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868
  1. """Exchange and Queue declarations."""
  2. from __future__ import absolute_import, unicode_literals
  3. import numbers
  4. from .abstract import MaybeChannelBound, Object
  5. from .exceptions import ContentDisallowed
  6. from .five import python_2_unicode_compatible, string_t
  7. from .serialization import prepare_accept_content
  8. TRANSIENT_DELIVERY_MODE = 1
  9. PERSISTENT_DELIVERY_MODE = 2
  10. DELIVERY_MODES = {'transient': TRANSIENT_DELIVERY_MODE,
  11. 'persistent': PERSISTENT_DELIVERY_MODE}
  12. __all__ = ('Exchange', 'Queue', 'binding', 'maybe_delivery_mode')
  13. INTERNAL_EXCHANGE_PREFIX = ('amq.',)
  14. def _reprstr(s):
  15. s = repr(s)
  16. if isinstance(s, string_t) and s.startswith("u'"):
  17. return s[2:-1]
  18. return s[1:-1]
  19. def pretty_bindings(bindings):
  20. return '[{0}]'.format(', '.join(map(str, bindings)))
  21. def maybe_delivery_mode(
  22. v, modes=DELIVERY_MODES, default=PERSISTENT_DELIVERY_MODE):
  23. """Get delivery mode by name (or none if undefined)."""
  24. if v:
  25. return v if isinstance(v, numbers.Integral) else modes[v]
  26. return default
  27. @python_2_unicode_compatible
  28. class Exchange(MaybeChannelBound):
  29. """An Exchange declaration.
  30. Arguments:
  31. name (str): See :attr:`name`.
  32. type (str): See :attr:`type`.
  33. channel (kombu.Connection, ChannelT): See :attr:`channel`.
  34. durable (bool): See :attr:`durable`.
  35. auto_delete (bool): See :attr:`auto_delete`.
  36. delivery_mode (enum): See :attr:`delivery_mode`.
  37. arguments (Dict): See :attr:`arguments`.
  38. no_declare (bool): See :attr:`no_declare`
  39. Attributes:
  40. name (str): Name of the exchange.
  41. Default is no name (the default exchange).
  42. type (str):
  43. *This description of AMQP exchange types was shamelessly stolen
  44. from the blog post `AMQP in 10 minutes: Part 4`_ by
  45. Rajith Attapattu. Reading this article is recommended if you're
  46. new to amqp.*
  47. "AMQP defines four default exchange types (routing algorithms) that
  48. covers most of the common messaging use cases. An AMQP broker can
  49. also define additional exchange types, so see your broker
  50. manual for more information about available exchange types.
  51. * `direct` (*default*)
  52. Direct match between the routing key in the message,
  53. and the routing criteria used when a queue is bound to
  54. this exchange.
  55. * `topic`
  56. Wildcard match between the routing key and the routing
  57. pattern specified in the exchange/queue binding.
  58. The routing key is treated as zero or more words delimited
  59. by `"."` and supports special wildcard characters. `"*"`
  60. matches a single word and `"#"` matches zero or more words.
  61. * `fanout`
  62. Queues are bound to this exchange with no arguments. Hence
  63. any message sent to this exchange will be forwarded to all
  64. queues bound to this exchange.
  65. * `headers`
  66. Queues are bound to this exchange with a table of arguments
  67. containing headers and values (optional). A special
  68. argument named "x-match" determines the matching algorithm,
  69. where `"all"` implies an `AND` (all pairs must match) and
  70. `"any"` implies `OR` (at least one pair must match).
  71. :attr:`arguments` is used to specify the arguments.
  72. .. _`AMQP in 10 minutes: Part 4`:
  73. https://bit.ly/2rcICv5
  74. channel (ChannelT): The channel the exchange is bound to (if bound).
  75. durable (bool): Durable exchanges remain active when a server restarts.
  76. Non-durable exchanges (transient exchanges) are purged when a
  77. server restarts. Default is :const:`True`.
  78. auto_delete (bool): If set, the exchange is deleted when all queues
  79. have finished using it. Default is :const:`False`.
  80. delivery_mode (enum): The default delivery mode used for messages.
  81. The value is an integer, or alias string.
  82. * 1 or `"transient"`
  83. The message is transient. Which means it is stored in
  84. memory only, and is lost if the server dies or restarts.
  85. * 2 or "persistent" (*default*)
  86. The message is persistent. Which means the message is
  87. stored both in-memory, and on disk, and therefore
  88. preserved if the server dies or restarts.
  89. The default value is 2 (persistent).
  90. arguments (Dict): Additional arguments to specify when the exchange
  91. is declared.
  92. no_declare (bool): Never declare this exchange
  93. (:meth:`declare` does nothing).
  94. """
  95. TRANSIENT_DELIVERY_MODE = TRANSIENT_DELIVERY_MODE
  96. PERSISTENT_DELIVERY_MODE = PERSISTENT_DELIVERY_MODE
  97. name = ''
  98. type = 'direct'
  99. durable = True
  100. auto_delete = False
  101. passive = False
  102. delivery_mode = None
  103. no_declare = False
  104. attrs = (
  105. ('name', None),
  106. ('type', None),
  107. ('arguments', None),
  108. ('durable', bool),
  109. ('passive', bool),
  110. ('auto_delete', bool),
  111. ('delivery_mode', lambda m: DELIVERY_MODES.get(m) or m),
  112. ('no_declare', bool),
  113. )
  114. def __init__(self, name='', type='', channel=None, **kwargs):
  115. super(Exchange, self).__init__(**kwargs)
  116. self.name = name or self.name
  117. self.type = type or self.type
  118. self.maybe_bind(channel)
  119. def __hash__(self):
  120. return hash('E|%s' % (self.name,))
  121. def _can_declare(self):
  122. return not self.no_declare and (
  123. self.name and not self.name.startswith(
  124. INTERNAL_EXCHANGE_PREFIX))
  125. def declare(self, nowait=False, passive=None, channel=None):
  126. """Declare the exchange.
  127. Creates the exchange on the broker, unless passive is set
  128. in which case it will only assert that the exchange exists.
  129. Argument:
  130. nowait (bool): If set the server will not respond, and a
  131. response will not be waited for. Default is :const:`False`.
  132. """
  133. if self._can_declare():
  134. passive = self.passive if passive is None else passive
  135. return (channel or self.channel).exchange_declare(
  136. exchange=self.name, type=self.type, durable=self.durable,
  137. auto_delete=self.auto_delete, arguments=self.arguments,
  138. nowait=nowait, passive=passive,
  139. )
  140. def bind_to(self, exchange='', routing_key='',
  141. arguments=None, nowait=False, channel=None, **kwargs):
  142. """Bind the exchange to another exchange.
  143. Arguments:
  144. nowait (bool): If set the server will not respond, and the call
  145. will not block waiting for a response.
  146. Default is :const:`False`.
  147. """
  148. if isinstance(exchange, Exchange):
  149. exchange = exchange.name
  150. return (channel or self.channel).exchange_bind(
  151. destination=self.name,
  152. source=exchange,
  153. routing_key=routing_key,
  154. nowait=nowait,
  155. arguments=arguments,
  156. )
  157. def unbind_from(self, source='', routing_key='',
  158. nowait=False, arguments=None, channel=None):
  159. """Delete previously created exchange binding from the server."""
  160. if isinstance(source, Exchange):
  161. source = source.name
  162. return (channel or self.channel).exchange_unbind(
  163. destination=self.name,
  164. source=source,
  165. routing_key=routing_key,
  166. nowait=nowait,
  167. arguments=arguments,
  168. )
  169. def Message(self, body, delivery_mode=None, properties=None, **kwargs):
  170. """Create message instance to be sent with :meth:`publish`.
  171. Arguments:
  172. body (Any): Message body.
  173. delivery_mode (bool): Set custom delivery mode.
  174. Defaults to :attr:`delivery_mode`.
  175. priority (int): Message priority, 0 to broker configured
  176. max priority, where higher is better.
  177. content_type (str): The messages content_type. If content_type
  178. is set, no serialization occurs as it is assumed this is either
  179. a binary object, or you've done your own serialization.
  180. Leave blank if using built-in serialization as our library
  181. properly sets content_type.
  182. content_encoding (str): The character set in which this object
  183. is encoded. Use "binary" if sending in raw binary objects.
  184. Leave blank if using built-in serialization as our library
  185. properly sets content_encoding.
  186. properties (Dict): Message properties.
  187. headers (Dict): Message headers.
  188. """
  189. # XXX This method is unused by kombu itself AFAICT [ask].
  190. properties = {} if properties is None else properties
  191. properties['delivery_mode'] = maybe_delivery_mode(self.delivery_mode)
  192. return self.channel.prepare_message(
  193. body,
  194. properties=properties,
  195. **kwargs)
  196. def publish(self, message, routing_key=None, mandatory=False,
  197. immediate=False, exchange=None):
  198. """Publish message.
  199. Arguments:
  200. message (Union[kombu.Message, str, bytes]):
  201. Message to publish.
  202. routing_key (str): Message routing key.
  203. mandatory (bool): Currently not supported.
  204. immediate (bool): Currently not supported.
  205. """
  206. if isinstance(message, string_t):
  207. message = self.Message(message)
  208. exchange = exchange or self.name
  209. return self.channel.basic_publish(
  210. message,
  211. exchange=exchange,
  212. routing_key=routing_key,
  213. mandatory=mandatory,
  214. immediate=immediate,
  215. )
  216. def delete(self, if_unused=False, nowait=False):
  217. """Delete the exchange declaration on server.
  218. Arguments:
  219. if_unused (bool): Delete only if the exchange has no bindings.
  220. Default is :const:`False`.
  221. nowait (bool): If set the server will not respond, and a
  222. response will not be waited for. Default is :const:`False`.
  223. """
  224. return self.channel.exchange_delete(exchange=self.name,
  225. if_unused=if_unused,
  226. nowait=nowait)
  227. def binding(self, routing_key='', arguments=None, unbind_arguments=None):
  228. return binding(self, routing_key, arguments, unbind_arguments)
  229. def __eq__(self, other):
  230. if isinstance(other, Exchange):
  231. return (self.name == other.name and
  232. self.type == other.type and
  233. self.arguments == other.arguments and
  234. self.durable == other.durable and
  235. self.auto_delete == other.auto_delete and
  236. self.delivery_mode == other.delivery_mode)
  237. return NotImplemented
  238. def __ne__(self, other):
  239. return not self.__eq__(other)
  240. def __repr__(self):
  241. return self._repr_entity(self)
  242. def __str__(self):
  243. return 'Exchange {0}({1})'.format(
  244. _reprstr(self.name) or repr(''), self.type,
  245. )
  246. @property
  247. def can_cache_declaration(self):
  248. return not self.auto_delete
  249. @python_2_unicode_compatible
  250. class binding(Object):
  251. """Represents a queue or exchange binding.
  252. Arguments:
  253. exchange (Exchange): Exchange to bind to.
  254. routing_key (str): Routing key used as binding key.
  255. arguments (Dict): Arguments for bind operation.
  256. unbind_arguments (Dict): Arguments for unbind operation.
  257. """
  258. attrs = (
  259. ('exchange', None),
  260. ('routing_key', None),
  261. ('arguments', None),
  262. ('unbind_arguments', None)
  263. )
  264. def __init__(self, exchange=None, routing_key='',
  265. arguments=None, unbind_arguments=None):
  266. self.exchange = exchange
  267. self.routing_key = routing_key
  268. self.arguments = arguments
  269. self.unbind_arguments = unbind_arguments
  270. def declare(self, channel, nowait=False):
  271. """Declare destination exchange."""
  272. if self.exchange and self.exchange.name:
  273. self.exchange.declare(channel=channel, nowait=nowait)
  274. def bind(self, entity, nowait=False, channel=None):
  275. """Bind entity to this binding."""
  276. entity.bind_to(exchange=self.exchange,
  277. routing_key=self.routing_key,
  278. arguments=self.arguments,
  279. nowait=nowait,
  280. channel=channel)
  281. def unbind(self, entity, nowait=False, channel=None):
  282. """Unbind entity from this binding."""
  283. entity.unbind_from(self.exchange,
  284. routing_key=self.routing_key,
  285. arguments=self.unbind_arguments,
  286. nowait=nowait,
  287. channel=channel)
  288. def __repr__(self):
  289. return '<binding: {0}>'.format(self)
  290. def __str__(self):
  291. return '{0}->{1}'.format(
  292. _reprstr(self.exchange.name), _reprstr(self.routing_key),
  293. )
  294. @python_2_unicode_compatible
  295. class Queue(MaybeChannelBound):
  296. """A Queue declaration.
  297. Arguments:
  298. name (str): See :attr:`name`.
  299. exchange (Exchange, str): See :attr:`exchange`.
  300. routing_key (str): See :attr:`routing_key`.
  301. channel (kombu.Connection, ChannelT): See :attr:`channel`.
  302. durable (bool): See :attr:`durable`.
  303. exclusive (bool): See :attr:`exclusive`.
  304. auto_delete (bool): See :attr:`auto_delete`.
  305. queue_arguments (Dict): See :attr:`queue_arguments`.
  306. binding_arguments (Dict): See :attr:`binding_arguments`.
  307. consumer_arguments (Dict): See :attr:`consumer_arguments`.
  308. no_declare (bool): See :attr:`no_declare`.
  309. on_declared (Callable): See :attr:`on_declared`.
  310. expires (float): See :attr:`expires`.
  311. message_ttl (float): See :attr:`message_ttl`.
  312. max_length (int): See :attr:`max_length`.
  313. max_length_bytes (int): See :attr:`max_length_bytes`.
  314. max_priority (int): See :attr:`max_priority`.
  315. Attributes:
  316. name (str): Name of the queue.
  317. Default is no name (default queue destination).
  318. exchange (Exchange): The :class:`Exchange` the queue binds to.
  319. routing_key (str): The routing key (if any), also called *binding key*.
  320. The interpretation of the routing key depends on
  321. the :attr:`Exchange.type`.
  322. * direct exchange
  323. Matches if the routing key property of the message and
  324. the :attr:`routing_key` attribute are identical.
  325. * fanout exchange
  326. Always matches, even if the binding does not have a key.
  327. * topic exchange
  328. Matches the routing key property of the message by a primitive
  329. pattern matching scheme. The message routing key then consists
  330. of words separated by dots (`"."`, like domain names), and
  331. two special characters are available; star (`"*"`) and hash
  332. (`"#"`). The star matches any word, and the hash matches
  333. zero or more words. For example `"*.stock.#"` matches the
  334. routing keys `"usd.stock"` and `"eur.stock.db"` but not
  335. `"stock.nasdaq"`.
  336. channel (ChannelT): The channel the Queue is bound to (if bound).
  337. durable (bool): Durable queues remain active when a server restarts.
  338. Non-durable queues (transient queues) are purged if/when
  339. a server restarts.
  340. Note that durable queues do not necessarily hold persistent
  341. messages, although it does not make sense to send
  342. persistent messages to a transient queue.
  343. Default is :const:`True`.
  344. exclusive (bool): Exclusive queues may only be consumed from by the
  345. current connection. Setting the 'exclusive' flag
  346. always implies 'auto-delete'.
  347. Default is :const:`False`.
  348. auto_delete (bool): If set, the queue is deleted when all consumers
  349. have finished using it. Last consumer can be canceled
  350. either explicitly or because its channel is closed. If
  351. there was no consumer ever on the queue, it won't be
  352. deleted.
  353. expires (float): Set the expiry time (in seconds) for when this
  354. queue should expire.
  355. The expiry time decides how long the queue can stay unused
  356. before it's automatically deleted.
  357. *Unused* means the queue has no consumers, the queue has not been
  358. redeclared, and ``Queue.get`` has not been invoked for a duration
  359. of at least the expiration period.
  360. See https://www.rabbitmq.com/ttl.html#queue-ttl
  361. **RabbitMQ extension**: Only available when using RabbitMQ.
  362. message_ttl (float): Message time to live in seconds.
  363. This setting controls how long messages can stay in the queue
  364. unconsumed. If the expiry time passes before a message consumer
  365. has received the message, the message is deleted and no consumer
  366. will see the message.
  367. See https://www.rabbitmq.com/ttl.html#per-queue-message-ttl
  368. **RabbitMQ extension**: Only available when using RabbitMQ.
  369. max_length (int): Set the maximum number of messages that the
  370. queue can hold.
  371. If the number of messages in the queue size exceeds this limit,
  372. new messages will be dropped (or dead-lettered if a dead letter
  373. exchange is active).
  374. See https://www.rabbitmq.com/maxlength.html
  375. **RabbitMQ extension**: Only available when using RabbitMQ.
  376. max_length_bytes (int): Set the max size (in bytes) for the total
  377. of messages in the queue.
  378. If the total size of all the messages in the queue exceeds this
  379. limit, new messages will be dropped (or dead-lettered if a dead
  380. letter exchange is active).
  381. **RabbitMQ extension**: Only available when using RabbitMQ.
  382. max_priority (int): Set the highest priority number for this queue.
  383. For example if the value is 10, then messages can delivered to
  384. this queue can have a ``priority`` value between 0 and 10,
  385. where 10 is the highest priority.
  386. RabbitMQ queues without a max priority set will ignore
  387. the priority field in the message, so if you want priorities
  388. you need to set the max priority field to declare the queue
  389. as a priority queue.
  390. **RabbitMQ extension**: Only available when using RabbitMQ.
  391. queue_arguments (Dict): Additional arguments used when declaring
  392. the queue. Can be used to to set the arguments value
  393. for RabbitMQ/AMQP's ``queue.declare``.
  394. binding_arguments (Dict): Additional arguments used when binding
  395. the queue. Can be used to to set the arguments value
  396. for RabbitMQ/AMQP's ``queue.declare``.
  397. consumer_arguments (Dict): Additional arguments used when consuming
  398. from this queue. Can be used to to set the arguments value
  399. for RabbitMQ/AMQP's ``basic.consume``.
  400. alias (str): Unused in Kombu, but applications can take advantage
  401. of this, for example to give alternate names to queues with
  402. automatically generated queue names.
  403. on_declared (Callable): Optional callback to be applied when the
  404. queue has been declared (the ``queue_declare`` operation is
  405. complete). This must be a function with a signature that
  406. accepts at least 3 positional arguments:
  407. ``(name, messages, consumers)``.
  408. no_declare (bool): Never declare this queue, nor related
  409. entities (:meth:`declare` does nothing).
  410. """
  411. ContentDisallowed = ContentDisallowed
  412. name = ''
  413. exchange = Exchange('')
  414. routing_key = ''
  415. durable = True
  416. exclusive = False
  417. auto_delete = False
  418. no_ack = False
  419. attrs = (
  420. ('name', None),
  421. ('exchange', None),
  422. ('routing_key', None),
  423. ('queue_arguments', None),
  424. ('binding_arguments', None),
  425. ('consumer_arguments', None),
  426. ('durable', bool),
  427. ('exclusive', bool),
  428. ('auto_delete', bool),
  429. ('no_ack', None),
  430. ('alias', None),
  431. ('bindings', list),
  432. ('no_declare', bool),
  433. ('expires', float),
  434. ('message_ttl', float),
  435. ('max_length', int),
  436. ('max_length_bytes', int),
  437. ('max_priority', int)
  438. )
  439. def __init__(self, name='', exchange=None, routing_key='',
  440. channel=None, bindings=None, on_declared=None,
  441. **kwargs):
  442. super(Queue, self).__init__(**kwargs)
  443. self.name = name or self.name
  444. if isinstance(exchange, str):
  445. self.exchange = Exchange(exchange)
  446. elif isinstance(exchange, Exchange):
  447. self.exchange = exchange
  448. self.routing_key = routing_key or self.routing_key
  449. self.bindings = set(bindings or [])
  450. self.on_declared = on_declared
  451. # allows Queue('name', [binding(...), binding(...), ...])
  452. if isinstance(exchange, (list, tuple, set)):
  453. self.bindings |= set(exchange)
  454. if self.bindings:
  455. self.exchange = None
  456. # exclusive implies auto-delete.
  457. if self.exclusive:
  458. self.auto_delete = True
  459. self.maybe_bind(channel)
  460. def bind(self, channel):
  461. on_declared = self.on_declared
  462. bound = super(Queue, self).bind(channel)
  463. bound.on_declared = on_declared
  464. return bound
  465. def __hash__(self):
  466. return hash('Q|%s' % (self.name,))
  467. def when_bound(self):
  468. if self.exchange:
  469. self.exchange = self.exchange(self.channel)
  470. def declare(self, nowait=False, channel=None):
  471. """Declare queue and exchange then binds queue to exchange."""
  472. if not self.no_declare:
  473. # - declare main binding.
  474. self._create_exchange(nowait=nowait, channel=channel)
  475. self._create_queue(nowait=nowait, channel=channel)
  476. self._create_bindings(nowait=nowait, channel=channel)
  477. return self.name
  478. def _create_exchange(self, nowait=False, channel=None):
  479. if self.exchange:
  480. self.exchange.declare(nowait=nowait, channel=channel)
  481. def _create_queue(self, nowait=False, channel=None):
  482. self.queue_declare(nowait=nowait, passive=False, channel=channel)
  483. if self.exchange and self.exchange.name:
  484. self.queue_bind(nowait=nowait, channel=channel)
  485. def _create_bindings(self, nowait=False, channel=None):
  486. for B in self.bindings:
  487. channel = channel or self.channel
  488. B.declare(channel)
  489. B.bind(self, nowait=nowait, channel=channel)
  490. def queue_declare(self, nowait=False, passive=False, channel=None):
  491. """Declare queue on the server.
  492. Arguments:
  493. nowait (bool): Do not wait for a reply.
  494. passive (bool): If set, the server will not create the queue.
  495. The client can use this to check whether a queue exists
  496. without modifying the server state.
  497. """
  498. channel = channel or self.channel
  499. queue_arguments = channel.prepare_queue_arguments(
  500. self.queue_arguments or {},
  501. expires=self.expires,
  502. message_ttl=self.message_ttl,
  503. max_length=self.max_length,
  504. max_length_bytes=self.max_length_bytes,
  505. max_priority=self.max_priority,
  506. )
  507. ret = channel.queue_declare(
  508. queue=self.name,
  509. passive=passive,
  510. durable=self.durable,
  511. exclusive=self.exclusive,
  512. auto_delete=self.auto_delete,
  513. arguments=queue_arguments,
  514. nowait=nowait,
  515. )
  516. if not self.name:
  517. self.name = ret[0]
  518. if self.on_declared:
  519. self.on_declared(*ret)
  520. return ret
  521. def queue_bind(self, nowait=False, channel=None):
  522. """Create the queue binding on the server."""
  523. return self.bind_to(self.exchange, self.routing_key,
  524. self.binding_arguments,
  525. channel=channel, nowait=nowait)
  526. def bind_to(self, exchange='', routing_key='',
  527. arguments=None, nowait=False, channel=None):
  528. if isinstance(exchange, Exchange):
  529. exchange = exchange.name
  530. return (channel or self.channel).queue_bind(
  531. queue=self.name,
  532. exchange=exchange,
  533. routing_key=routing_key,
  534. arguments=arguments,
  535. nowait=nowait,
  536. )
  537. def get(self, no_ack=None, accept=None):
  538. """Poll the server for a new message.
  539. This method provides direct access to the messages in a
  540. queue using a synchronous dialogue, designed for
  541. specific types of applications where synchronous functionality
  542. is more important than performance.
  543. Returns:
  544. ~kombu.Message: if a message was available,
  545. or :const:`None` otherwise.
  546. Arguments:
  547. no_ack (bool): If enabled the broker will
  548. automatically ack messages.
  549. accept (Set[str]): Custom list of accepted content types.
  550. """
  551. no_ack = self.no_ack if no_ack is None else no_ack
  552. message = self.channel.basic_get(queue=self.name, no_ack=no_ack)
  553. if message is not None:
  554. m2p = getattr(self.channel, 'message_to_python', None)
  555. if m2p:
  556. message = m2p(message)
  557. if message.errors:
  558. message._reraise_error()
  559. message.accept = prepare_accept_content(accept)
  560. return message
  561. def purge(self, nowait=False):
  562. """Remove all ready messages from the queue."""
  563. return self.channel.queue_purge(queue=self.name,
  564. nowait=nowait) or 0
  565. def consume(self, consumer_tag='', callback=None,
  566. no_ack=None, nowait=False):
  567. """Start a queue consumer.
  568. Consumers last as long as the channel they were created on, or
  569. until the client cancels them.
  570. Arguments:
  571. consumer_tag (str): Unique identifier for the consumer.
  572. The consumer tag is local to a connection, so two clients
  573. can use the same consumer tags. If this field is empty
  574. the server will generate a unique tag.
  575. no_ack (bool): If enabled the broker will automatically
  576. ack messages.
  577. nowait (bool): Do not wait for a reply.
  578. callback (Callable): callback called for each delivered message.
  579. """
  580. if no_ack is None:
  581. no_ack = self.no_ack
  582. return self.channel.basic_consume(
  583. queue=self.name,
  584. no_ack=no_ack,
  585. consumer_tag=consumer_tag or '',
  586. callback=callback,
  587. nowait=nowait,
  588. arguments=self.consumer_arguments)
  589. def cancel(self, consumer_tag):
  590. """Cancel a consumer by consumer tag."""
  591. return self.channel.basic_cancel(consumer_tag)
  592. def delete(self, if_unused=False, if_empty=False, nowait=False):
  593. """Delete the queue.
  594. Arguments:
  595. if_unused (bool): If set, the server will only delete the queue
  596. if it has no consumers. A channel error will be raised
  597. if the queue has consumers.
  598. if_empty (bool): If set, the server will only delete the queue if
  599. it is empty. If it is not empty a channel error will be raised.
  600. nowait (bool): Do not wait for a reply.
  601. """
  602. return self.channel.queue_delete(queue=self.name,
  603. if_unused=if_unused,
  604. if_empty=if_empty,
  605. nowait=nowait)
  606. def queue_unbind(self, arguments=None, nowait=False, channel=None):
  607. return self.unbind_from(self.exchange, self.routing_key,
  608. arguments, nowait, channel)
  609. def unbind_from(self, exchange='', routing_key='',
  610. arguments=None, nowait=False, channel=None):
  611. """Unbind queue by deleting the binding from the server."""
  612. return (channel or self.channel).queue_unbind(
  613. queue=self.name,
  614. exchange=exchange.name,
  615. routing_key=routing_key,
  616. arguments=arguments,
  617. nowait=nowait,
  618. )
  619. def __eq__(self, other):
  620. if isinstance(other, Queue):
  621. return (self.name == other.name and
  622. self.exchange == other.exchange and
  623. self.routing_key == other.routing_key and
  624. self.queue_arguments == other.queue_arguments and
  625. self.binding_arguments == other.binding_arguments and
  626. self.consumer_arguments == other.consumer_arguments and
  627. self.durable == other.durable and
  628. self.exclusive == other.exclusive and
  629. self.auto_delete == other.auto_delete)
  630. return NotImplemented
  631. def __ne__(self, other):
  632. return not self.__eq__(other)
  633. def __repr__(self):
  634. if self.bindings:
  635. return self._repr_entity('Queue {name} -> {bindings}'.format(
  636. name=_reprstr(self.name),
  637. bindings=pretty_bindings(self.bindings),
  638. ))
  639. return self._repr_entity(
  640. 'Queue {name} -> {0.exchange!r} -> {routing_key}'.format(
  641. self, name=_reprstr(self.name),
  642. routing_key=_reprstr(self.routing_key),
  643. ),
  644. )
  645. @property
  646. def can_cache_declaration(self):
  647. if self.queue_arguments:
  648. expiring_queue = "x-expires" in self.queue_arguments
  649. else:
  650. expiring_queue = False
  651. return not expiring_queue and not self.auto_delete
  652. @classmethod
  653. def from_dict(cls, queue, **options):
  654. binding_key = options.get('binding_key') or options.get('routing_key')
  655. e_durable = options.get('exchange_durable')
  656. if e_durable is None:
  657. e_durable = options.get('durable')
  658. e_auto_delete = options.get('exchange_auto_delete')
  659. if e_auto_delete is None:
  660. e_auto_delete = options.get('auto_delete')
  661. q_durable = options.get('queue_durable')
  662. if q_durable is None:
  663. q_durable = options.get('durable')
  664. q_auto_delete = options.get('queue_auto_delete')
  665. if q_auto_delete is None:
  666. q_auto_delete = options.get('auto_delete')
  667. e_arguments = options.get('exchange_arguments')
  668. q_arguments = options.get('queue_arguments')
  669. b_arguments = options.get('binding_arguments')
  670. c_arguments = options.get('consumer_arguments')
  671. bindings = options.get('bindings')
  672. exchange = Exchange(options.get('exchange'),
  673. type=options.get('exchange_type'),
  674. delivery_mode=options.get('delivery_mode'),
  675. routing_key=options.get('routing_key'),
  676. durable=e_durable,
  677. auto_delete=e_auto_delete,
  678. arguments=e_arguments)
  679. return Queue(queue,
  680. exchange=exchange,
  681. routing_key=binding_key,
  682. durable=q_durable,
  683. exclusive=options.get('exclusive'),
  684. auto_delete=q_auto_delete,
  685. no_ack=options.get('no_ack'),
  686. queue_arguments=q_arguments,
  687. binding_arguments=b_arguments,
  688. consumer_arguments=c_arguments,
  689. bindings=bindings)
  690. def as_dict(self, recurse=False):
  691. res = super(Queue, self).as_dict(recurse)
  692. if not recurse:
  693. return res
  694. bindings = res.get('bindings')
  695. if bindings:
  696. res['bindings'] = [b.as_dict(recurse=True) for b in bindings]
  697. return res