Skip to content

Commit a888812

Browse files
coseto6125coseto6125
committed
feat(event-topic): RabbitMQ Python detector (T5-8)
Adds `RABBITMQ_PYTHON` `EventTopicConfig` covering pika (sync AMQP), aio_pika (async AMQP), and kombu. Uses the already-shipped `classify_amqp_direction` (T5-1) to distinguish Publish from Subscribe call sites — making this the first config in the T5 family to exercise both directions in the same PR. Topic literal semantics: `routing_key` kwarg for publish calls (AMQP routing unit), `queue` kwarg for pika basic_consume (subscriber binding point). The topology distinction (exchange type, binding key, queue-vs- exchange namespace) is not expressible in the current `RawEventTopic` schema; deferred to schema-migration PR — see rabbitmq_python.rs module doc for the full gap description and the concrete LLM-query example that requires `kind: Option<StrRef>`. - `crates/ecp-analyzer/src/event_topic/rabbitmq_python.rs` — new `RABBITMQ_PYTHON` const; full schema-gap doc including deferred `kind` field rationale - `crates/ecp-analyzer/src/event_topic/mod.rs` — expose `pub mod rabbitmq_python` + re-export `RABBITMQ_PYTHON` - `crates/ecp-analyzer/src/python/frameworks.scm` — append RabbitMQ section: pika basic_publish/basic_consume (kwarg patterns), aio_pika async exchange.publish, kombu producer.publish; `amqp.direction` capture binds method name for classifier - `crates/ecp-analyzer/src/python/parser.rs` — extend config slice to `&[KAFKA_PYTHON, RABBITMQ_PYTHON]` - `crates/ecp-analyzer/tests/python_rabbitmq_events.rs` — 8 tests: pika publish, pika consume, aio_pika publish, kombu publish, variable routing_key emits nothing, no-import gate, publish+consume same file, kafka-import isolation
1 parent c682239 commit a888812

5 files changed

Lines changed: 381 additions & 1 deletion

File tree

‎crates/ecp-analyzer/src/event_topic/mod.rs‎

Lines changed: 2 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -2,7 +2,9 @@ pub mod config;
22
pub mod extract;
33
pub mod kafka_python;
44
pub mod normalize;
5+
pub mod rabbitmq_python;
56

67
pub use config::EventTopicConfig;
78
pub use extract::{classify_amqp_direction, classify_kafka_direction, extract_event_topics};
89
pub use kafka_python::KAFKA_PYTHON;
10+
pub use rabbitmq_python::RABBITMQ_PYTHON;

‎crates/ecp-analyzer/src/event_topic/rabbitmq_python.rs‎

Lines changed: 62 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -0,0 +1,62 @@
1+
//! `EventTopicConfig` for RabbitMQ / AMQP Python clients.
2+
//!
3+
//! Covers three Python RabbitMQ libraries under a single config:
4+
//! - `pika` (sync AMQP 0-9-1): `channel.basic_publish(routing_key=...)` / `channel.basic_consume(queue=...)`
5+
//! - `aio_pika` (async AMQP 0-9-1): `exchange.publish(..., routing_key=...)` / `queue.consume(callback)`
6+
//! - `kombu` (abstraction layer): `producer.publish(body, routing_key=...)`
7+
//!
8+
//! Direction dispatch: `classify_amqp_direction` (T5-1) maps call-method text
9+
//! to `PubSub::Publish` / `PubSub::Subscribe` — the `direction_capture` slot
10+
//! binds the method identifier so the classifier sees `"basic_publish"`,
11+
//! `"basic_consume"`, `"publish"`, or `"consume"`.
12+
//!
13+
//! # Topic literal semantics
14+
//! - Publish: `routing_key` kwarg value (AMQP's canonical routing unit).
15+
//! - Subscribe: `queue` kwarg / positional arg value (what the subscriber binds to).
16+
//!
17+
//! Both are stored in `RawEventTopic::topic_literal`. The queue/exchange topology
18+
//! distinction (direct vs topic vs fanout, exchange name, binding key) cannot be
19+
//! expressed in the current `RawEventTopic` schema — see **Schema gap** below.
20+
//!
21+
//! # Schema gap (deferred to schema-migration PR)
22+
//! `RawEventTopic` has no `kind` field. For RabbitMQ this loses:
23+
//! - Whether the topic string is a routing_key, a queue name, or an exchange name.
24+
//! - The exchange type (`direct` / `topic` / `fanout` / `headers`) — relevant because
25+
//! fanout ignores routing_key entirely, so the stored literal would be meaningless.
26+
//! - Binding topology: a single queue can be bound to multiple routing patterns.
27+
//!
28+
//! Concrete LLM-query example that the missing field blocks:
29+
//! "Find all publishers whose routing_key matches subscriber queues bound to
30+
//! the `orders.direct` exchange" — without a `kind: Option<StrRef>` field,
31+
//! the graph cannot distinguish publish-side routing_key strings from
32+
//! subscribe-side queue names in the same `EventTopic` node set, forcing
33+
//! the LLM to guess topology from naming conventions rather than graph edges.
34+
//!
35+
//! The fix is `RawEventTopic { kind: Option<StrRef> }` added append-only, with
36+
//! values `"routing_key"`, `"queue"`, `"exchange"` — this PR intentionally defers
37+
//! that migration so the schema change ships in one coordinated PR.
38+
//!
39+
//! # LLM-utility justification (graph-completeness criterion A)
40+
//! Without this config, `ecp impact` is blind to RabbitMQ message paths. A
41+
//! refactor of a `publish_order` function that emits to `routing_key='orders'`
42+
//! would show zero downstream consumers, causing the LLM to declare the change
43+
//! safe when it actually breaks every order-processing service bound to that queue.
44+
45+
use super::config::EventTopicConfig;
46+
use super::extract::classify_amqp_direction;
47+
use ecp_core::analyzer::types::FrameworkId;
48+
49+
/// RabbitMQ Python detector — fires for `pika`, `aio_pika`, and `kombu` imports.
50+
///
51+
/// `direction_capture: "amqp.direction"` binds the call-method identifier
52+
/// (e.g. `basic_publish`, `basic_consume`, `publish`, `consume`) so that
53+
/// `classify_amqp_direction` can resolve `PubSub` direction without fabrication.
54+
pub const RABBITMQ_PYTHON: EventTopicConfig = EventTopicConfig {
55+
framework: FrameworkId::RabbitMq,
56+
topic_capture: "amqp.topic",
57+
producer_capture: "amqp.fn",
58+
direction_capture: "amqp.direction",
59+
import_gate: &["pika", "aio_pika", "kombu"],
60+
direction_classifier: classify_amqp_direction,
61+
canonicalize: true,
62+
};

‎crates/ecp-analyzer/src/python/frameworks.scm‎

Lines changed: 82 additions & 0 deletions
Original file line numberDiff line numberDiff line change
@@ -239,6 +239,88 @@
239239
arguments: (argument_list
240240
. (string) @kafka.topic)))))
241241

242+
;; ---- RabbitMQ Python (T5-8) ----
243+
;; Covers pika (sync AMQP), aio_pika (async AMQP), and kombu.
244+
;; Import gate (`pika`, `aio_pika`, `kombu`) is enforced by
245+
;; RABBITMQ_PYTHON.import_gate — these queries fire on syntax alone.
246+
;;
247+
;; `amqp.direction` binds the method identifier so `classify_amqp_direction`
248+
;; can resolve Publish vs Subscribe direction.
249+
;;
250+
;; Topic literal semantics (documented in rabbitmq_python.rs):
251+
;; Publish → `routing_key` kwarg value (AMQP routing unit).
252+
;; Subscribe → `queue` kwarg / positional value (subscriber binding point).
253+
;; The queue/exchange topology distinction cannot be expressed in the current
254+
;; `RawEventTopic` schema; deferred to schema-migration PR.
255+
;;
256+
;; Anchored to `function_definition` to co-capture enclosing function name.
257+
;; Module-level calls are omitted — same rationale as Kafka (T5-2).
258+
259+
;; pika (sync): channel.basic_publish(exchange=..., routing_key='orders', body=...)
260+
;; Captures routing_key kwarg string literal as topic.
261+
(function_definition
262+
name: (identifier) @amqp.fn
263+
body: (block
264+
(_
265+
(call
266+
function: (attribute
267+
attribute: (identifier) @amqp.direction (#eq? @amqp.direction "basic_publish"))
268+
arguments: (argument_list
269+
(keyword_argument
270+
name: (identifier) @_rk (#eq? @_rk "routing_key")
271+
value: (string) @amqp.topic))))))
272+
273+
;; pika (sync): channel.basic_consume(queue='orders', on_message_callback=cb)
274+
;; Captures queue kwarg string literal as topic.
275+
(function_definition
276+
name: (identifier) @amqp.fn
277+
body: (block
278+
(_
279+
(call
280+
function: (attribute
281+
attribute: (identifier) @amqp.direction (#eq? @amqp.direction "basic_consume"))
282+
arguments: (argument_list
283+
(keyword_argument
284+
name: (identifier) @_q (#eq? @_q "queue")
285+
value: (string) @amqp.topic))))))
286+
287+
;; aio_pika (async): await exchange.publish(message, routing_key='orders')
288+
;; Captures routing_key kwarg string literal as topic.
289+
(function_definition
290+
name: (identifier) @amqp.fn
291+
body: (block
292+
(_
293+
(await
294+
(call
295+
function: (attribute
296+
attribute: (identifier) @amqp.direction (#eq? @amqp.direction "publish"))
297+
arguments: (argument_list
298+
(keyword_argument
299+
name: (identifier) @_rk2 (#eq? @_rk2 "routing_key")
300+
value: (string) @amqp.topic)))))))
301+
302+
;; aio_pika (async): await queue.consume(callback)
303+
;; Positional first arg is the callback — topic is the queue object itself.
304+
;; We cannot capture a string literal from the queue variable, so we anchor
305+
;; on the method name only; topic string must come from queue binding context.
306+
;; Pattern omitted: queue.consume() takes no string literal argument — topic
307+
;; resolution requires queue-name lookup which is out-of-scope for literal-only capture.
308+
;; Documented as T5-8-followup: needs `kind` field + queue-name resolver.
309+
310+
;; kombu: producer.publish(body, routing_key='orders', exchange='x')
311+
;; Captures routing_key kwarg string literal as topic.
312+
(function_definition
313+
name: (identifier) @amqp.fn
314+
body: (block
315+
(_
316+
(call
317+
function: (attribute
318+
attribute: (identifier) @amqp.direction (#eq? @amqp.direction "publish"))
319+
arguments: (argument_list
320+
(keyword_argument
321+
name: (identifier) @_rk3 (#eq? @_rk3 "routing_key")
322+
value: (string) @amqp.topic))))))
323+
242324
;; ---- SQLAlchemy declarative ORM (T4-3) ----
243325
;; Idiom A — classic Column() declarative (1.x and 2.x compatible).
244326
;; Captures: owner class name, field identifier, first positional arg of Column()

‎crates/ecp-analyzer/src/python/parser.rs‎

Lines changed: 4 additions & 1 deletion
Original file line numberDiff line numberDiff line change
@@ -1102,7 +1102,10 @@ impl LanguageProvider for PythonProvider {
11021102
&tree,
11031103
source,
11041104
&self.query,
1105-
&[crate::event_topic::KAFKA_PYTHON],
1105+
&[
1106+
crate::event_topic::KAFKA_PYTHON,
1107+
crate::event_topic::RABBITMQ_PYTHON,
1108+
],
11061109
&imports,
11071110
&mut pool,
11081111
);

0 commit comments

Comments
 (0)