AOP Framework
Cross-cutting concerns without touching business code. Build a method interception system like Spring AOP where logging, timing, caching, and retry wrap any method call through composable aspects.
Key Abstractions
Cross-cutting concern interface -- defines before, after, or around advice logic
Concrete aspects — each hooks into the around-advice lifecycle (before + invoke + after)
Predicate selecting which methods on which targets get intercepted
Chain of responsibility that stacks multiple aspects and invokes them in order
Creates proxy objects that wrap real targets and route calls through the aspect chain
Runtime context about the intercepted call: target, method name, arguments, return value
Class Diagram
How It Works
Logging, timing, caching, retries. These concerns show up everywhere in a codebase, but they have nothing to do with business logic. If you sprinkle logger.info(...) calls inside every service method, you end up with tangled code that is hard to change. Want to add timing? Touch every method. Want to remove logging from one module? Hunt through dozens of files.
Aspect-Oriented Programming separates these cross-cutting concerns from the code they apply to. You write each concern once as an Aspect, define a Pointcut that says which methods it applies to, and the framework weaves them in at runtime through Proxies. Your business classes stay clean. They don't import a logging library. They don't know they are being timed. They just do their job.
The mechanism is straightforward. When you ask the ProxyFactory for a proxy around your Calculator, it hands back an object that looks and behaves like a Calculator. But every method call passes through an AspectChain first. The chain walks through your registered aspects in order. Each aspect can run code before the call, after the call, or wrap the entire call with around advice. The around advice receives a proceed callable. Calling proceed() moves to the next aspect in the chain. The last proceed calls the real method on the real object. If an aspect never calls proceed, the real method never runs. That is how caching works: on a cache hit, you return the cached value and skip proceed entirely.
In Python, __getattr__ on the proxy intercepts attribute lookups. When the caller accesses a method, the proxy checks the pointcut. If it matches, the proxy returns a wrapper function that runs the chain. If it does not match, the call goes straight to the target. In Java, java.lang.reflect.Proxy with an InvocationHandler does the same thing for interface-based proxies.
Requirements
Functional
- Define aspects with
before,after, andaroundadvice hooks - Pointcuts control which methods on a target get intercepted
- Multiple aspects stack through a chain of responsibility, executing in registration order
- Around advice controls the full call lifecycle: inspect args, proceed or skip, modify return values
- Proxies are transparent to callers: same interface, same method signatures
Non-Functional
- Zero changes to target classes: the
Calculatorclass has no awareness of AOP - Composable: adding or removing an aspect is a configuration change, not a code change
- Predictable ordering: aspects execute in the order they are registered, no surprises
Design Decisions
Why use proxies instead of modifying the target class directly?
Modifying the target class (through monkey-patching, subclassing, or bytecode rewriting) couples the cross-cutting logic to the business class. If you monkey-patch Calculator.add to add logging, every user of that Calculator gets logging whether they want it or not. Proxies keep the original class untouched. You create a proxied version when you need aspects and use the original when you don't. Different parts of the application can have different aspect configurations for the same class. Testing is simpler too: test the raw class without any interception.
How should aspects compose when you stack three or four of them?
Chain of Responsibility. Each aspect wraps the next one. If you register [Logging, Timing, Caching], the call flow is: Logging.before -> Logging.around calls proceed -> Timing.around calls proceed -> Caching.around (maybe calls proceed, maybe returns cached value) -> real method. The chain builds recursively from the inside out. The outermost aspect sees everything, including time spent in inner aspects. This is the same pattern as middleware in web frameworks: Express, Django, Gin. They all use the same nested-proceed model.
Why separate before/after from around advice?
Most cross-cutting concerns only need a hook before or after the call. Logging, for example, just wants to print before and after. Writing a full around method for that is overkill. By offering before and after as simple hooks alongside around, you keep simple aspects simple. The framework calls before, then around (which calls proceed by default), then after. Only aspects that need to control flow, like caching or retry, override around. Spring AOP follows the same split: @Before, @After, @Around are separate annotation types.
What about the self-invocation problem?
If Calculator.add() internally calls this.divide(), that call goes to the real object, not the proxy. The proxy only wraps external calls. So the aspects never see divide(). This is a well-known limitation of proxy-based AOP. Spring documents it prominently. Two workarounds: (1) inject the proxy into the bean so internal calls go through the proxy (ugly but works), or (2) use bytecode weaving (AspectJ's compile-time or load-time weaving) instead of proxies. For an interview, knowing this limitation and being able to explain why it happens is more important than solving it.
Interview Follow-ups
- "How would you support annotation-based pointcuts?" Instead of matching on method names, inspect method metadata. In Python, you can attach custom attributes to functions (e.g.,
@cacheablesetsfunc._cacheable = True). The pointcut checks for that attribute. In Java, define a custom annotation like@Loggableand have the pointcut callmethod.isAnnotationPresent(Loggable.class). This is exactly how Spring's@Transactionalworks: the transaction aspect's pointcut matches methods annotated with@Transactional. - "How does Spring AOP handle aspect ordering?" Spring uses the
@Orderannotation or theOrderedinterface. Lower values run first (closer to the caller). When two aspects have the same order, the behavior is undefined, which is why Spring's docs tell you to always specify order explicitly. In your implementation, you could add aget_order()method to theAspectinterface and sort the aspect list before building the chain. - "What is the difference between Spring AOP and AspectJ?" Spring AOP is proxy-based and only intercepts calls made through the proxy. It works at runtime and only supports method-level join points. AspectJ is a full AOP framework that modifies bytecode at compile time or load time. It can intercept field access, constructor calls, static methods, and self-invocations. AspectJ is more powerful but adds build complexity. Spring defaults to its own proxy-based approach and lets you opt in to AspectJ when you need the extra power.
- "How would you add an async aspect that offloads work to a thread pool?" The
aroundadvice would submit theproceedcall to a thread pool and return aFuture(orCompletableFuturein Java,asyncio.Futurein Python). The caller gets the future immediately and awaits the result when ready. The tricky part is that other aspects in the chain might not expect async returns, so you need to decide whether async wrapping happens at the outermost layer only or if the entire chain becomes async. Spring's@Asyncannotation works this way: the proxy submits the call to aTaskExecutorand returns aFuture.
Code Implementation
1 from __future__ import annotations
2 from abc import ABC, abstractmethod
3 from dataclasses import dataclass, field
4 from typing import Any, Callable, List, Optional
5 import time
6 import functools
7
8
9 # --------------- JoinPoint ---------------
10
11 @dataclass
12 class JoinPoint:
13 """Runtime context for the intercepted method call."""
14 target: Any
15 method_name: str
16 args: tuple
17 kwargs: dict
18 return_value: Any = None
19 exception: Optional[Exception] = None
20
21
22 # --------------- Aspect Interface ---------------
23
24 class Aspect(ABC):
25 """
26 Base class for all aspects. Override the hooks you need.
27 around() gets a proceed callable that invokes the next aspect or the
28 real method.
29 """
30
31 def before(self, joinpoint: JoinPoint) -> None:
32 pass
33
34 def after(self, joinpoint: JoinPoint) -> None:
35 pass
36
37 def around(self, joinpoint: JoinPoint, proceed: Callable) -> Any:
38 return proceed()
39
40
41 # --------------- Pointcut ---------------
42
43 class Pointcut(ABC):
44 @abstractmethod
45 def matches(self, method_name: str) -> bool: ...
46
47
48 class AllMethodsPointcut(Pointcut):
49 """Matches every public method (no underscore prefix)."""
50 def matches(self, method_name: str) -> bool:
51 return not method_name.startswith("_")
52
53
54 class NamePatternPointcut(Pointcut):
55 """Matches methods whose names contain the given substring."""
56 def __init__(self, pattern: str):
57 self._pattern = pattern
58
59 def matches(self, method_name: str) -> bool:
60 return self._pattern in method_name
61
62
63 # --------------- Concrete Aspects ---------------
64
65 class LoggingAspect(Aspect):
66 def before(self, joinpoint: JoinPoint) -> None:
67 args_str = ", ".join(str(a) for a in joinpoint.args)
68 print(f" [LOG] Calling {joinpoint.method_name}({args_str})")
69
70 def after(self, joinpoint: JoinPoint) -> None:
71 if joinpoint.exception:
72 print(f" [LOG] {joinpoint.method_name} raised {joinpoint.exception}")
73 else:
74 print(f" [LOG] {joinpoint.method_name} returned {joinpoint.return_value}")
75
76
77 class TimingAspect(Aspect):
78 def around(self, joinpoint: JoinPoint, proceed: Callable) -> Any:
79 start = time.perf_counter()
80 result = proceed()
81 elapsed_ms = (time.perf_counter() - start) * 1000
82 print(f" [TIMER] {joinpoint.method_name} took {elapsed_ms:.2f}ms")
83 return result
84
85
86 class RetryAspect(Aspect):
87 def __init__(self, max_retries: int = 3, delay: float = 0.1):
88 self._max_retries = max_retries
89 self._delay = delay
90
91 def around(self, joinpoint: JoinPoint, proceed: Callable) -> Any:
92 last_error = None
93 for attempt in range(1, self._max_retries + 1):
94 try:
95 return proceed()
96 except Exception as e:
97 last_error = e
98 print(f" [RETRY] {joinpoint.method_name} attempt {attempt} failed: {e}")
99 if attempt < self._max_retries:
100 time.sleep(self._delay)
101 raise last_error
102
103
104 class CachingAspect(Aspect):
105 def __init__(self):
106 self._cache: dict[str, Any] = {}
107
108 def around(self, joinpoint: JoinPoint, proceed: Callable) -> Any:
109 cache_key = f"{joinpoint.method_name}:{joinpoint.args}"
110 if cache_key in self._cache:
111 print(f" [CACHE] Hit for {cache_key}")
112 return self._cache[cache_key]
113 result = proceed()
114 self._cache[cache_key] = result
115 print(f" [CACHE] Stored result for {cache_key}")
116 return result
117
118
119 # --------------- Aspect Chain ---------------
120
121 class AspectChain:
122 """
123 Chains aspects together using the Chain of Responsibility pattern.
124 Each aspect's around() receives a proceed callable that moves to the
125 next aspect in the chain, and the final proceed calls the real method.
126 """
127
128 def __init__(self, target: Any, aspects: List[Aspect]):
129 self._target = target
130 self._aspects = aspects
131
132 def invoke(self, method_name: str, args: tuple, kwargs: dict) -> Any:
133 joinpoint = JoinPoint(
134 target=self._target,
135 method_name=method_name,
136 args=args,
137 kwargs=kwargs,
138 )
139
140 # Build the chain from inside out.
141 # The innermost callable runs the real method.
142 real_method = getattr(self._target, method_name)
143
144 def make_proceed(index: int) -> Callable:
145 if index >= len(self._aspects):
146 # End of chain: call the actual method
147 def call_real():
148 return real_method(*args, **kwargs)
149 return call_real
150 else:
151 aspect = self._aspects[index]
152 next_proceed = make_proceed(index + 1)
153
154 def call_aspect():
155 aspect.before(joinpoint)
156 try:
157 result = aspect.around(joinpoint, next_proceed)
158 joinpoint.return_value = result
159 aspect.after(joinpoint)
160 return result
161 except Exception as e:
162 joinpoint.exception = e
163 aspect.after(joinpoint)
164 raise
165
166 return call_aspect
167
168 return make_proceed(0)()
169
170
171 # --------------- Proxy Factory ---------------
172
173 class ProxyFactory:
174 """Creates proxy objects that intercept method calls through the aspect chain."""
175
176 @staticmethod
177 def create_proxy(
178 target: Any,
179 aspects: List[Aspect],
180 pointcut: Pointcut,
181 ) -> Any:
182 return AopProxy(target, aspects, pointcut)
183
184
185 class AopProxy:
186 """
187 Proxy that intercepts attribute access. For methods matched by the
188 pointcut, calls go through the AspectChain. Everything else delegates
189 directly to the target.
190 """
191
192 def __init__(self, target: Any, aspects: List[Aspect], pointcut: Pointcut):
193 # Use object.__setattr__ to avoid triggering __getattr__
194 object.__setattr__(self, "_target", target)
195 object.__setattr__(self, "_chain", AspectChain(target, aspects))
196 object.__setattr__(self, "_pointcut", pointcut)
197
198 def __getattr__(self, name: str) -> Any:
199 attr = getattr(self._target, name)
200 if callable(attr) and self._pointcut.matches(name):
201 @functools.wraps(attr)
202 def intercepted(*args, **kwargs):
203 return self._chain.invoke(name, args, kwargs)
204 return intercepted
205 return attr
206
207
208 # --------------- Example Target Class ---------------
209
210 class Calculator:
211 """Plain business class. Knows nothing about aspects."""
212
213 def add(self, a: float, b: float) -> float:
214 return a + b
215
216 def divide(self, a: float, b: float) -> float:
217 if b == 0:
218 raise ValueError("Cannot divide by zero")
219 return a / b
220
221 def slow_multiply(self, a: float, b: float) -> float:
222 time.sleep(0.05) # Simulate slow computation
223 return a * b
224
225
226 # --------------- Demo ---------------
227
228 if __name__ == "__main__":
229
230 print("=== Create Calculator with Logging + Timing Aspects ===")
231 calc = Calculator()
232 proxy = ProxyFactory.create_proxy(
233 target=calc,
234 aspects=[LoggingAspect(), TimingAspect()],
235 pointcut=AllMethodsPointcut(),
236 )
237
238 print("\n--- add(3, 4) ---")
239 result = proxy.add(3, 4)
240 print(f" Result: {result}")
241
242 print("\n--- slow_multiply(5, 6) ---")
243 result = proxy.slow_multiply(5, 6)
244 print(f" Result: {result}")
245
246 print("\n--- divide(10, 0) with error ---")
247 try:
248 proxy.divide(10, 0)
249 except ValueError as e:
250 print(f" Caught expected error: {e}")
251
252 print("\n=== Caching Aspect ===")
253 cached_proxy = ProxyFactory.create_proxy(
254 target=calc,
255 aspects=[LoggingAspect(), CachingAspect()],
256 pointcut=AllMethodsPointcut(),
257 )
258 print("\n--- First call: add(10, 20) ---")
259 print(f" Result: {cached_proxy.add(10, 20)}")
260 print("\n--- Second call: add(10, 20) (should hit cache) ---")
261 print(f" Result: {cached_proxy.add(10, 20)}")
262
263 print("\n=== Retry Aspect ===")
264
265 class FlakyService:
266 def __init__(self):
267 self._call_count = 0
268
269 def fetch_data(self) -> str:
270 self._call_count += 1
271 if self._call_count < 3:
272 raise ConnectionError(f"Network error (attempt {self._call_count})")
273 return "data from remote service"
274
275 flaky = FlakyService()
276 retry_proxy = ProxyFactory.create_proxy(
277 target=flaky,
278 aspects=[LoggingAspect(), RetryAspect(max_retries=3, delay=0.05)],
279 pointcut=AllMethodsPointcut(),
280 )
281 print("\n--- fetch_data() with retries ---")
282 result = retry_proxy.fetch_data()
283 print(f" Result: {result}")
284
285 print("\n=== Selective Pointcut (only 'add' methods) ===")
286 selective_proxy = ProxyFactory.create_proxy(
287 target=calc,
288 aspects=[LoggingAspect()],
289 pointcut=NamePatternPointcut("add"),
290 )
291 print("\n--- add(1, 2) matches pointcut ---")
292 print(f" Result: {selective_proxy.add(1, 2)}")
293 print("\n--- divide(9, 3) does NOT match pointcut (no interception) ---")
294 print(f" Result: {selective_proxy.divide(9, 3)}")
295
296 print("\n=== Stacking Order Matters ===")
297 print(" Order: [Timing, Logging] -- timer wraps everything including log overhead")
298 proxy_a = ProxyFactory.create_proxy(calc, [TimingAspect(), LoggingAspect()], AllMethodsPointcut())
299 proxy_a.add(1, 1)
300
301 print("\n Order: [Logging, Timing] -- log fires first, then timer measures just the call")
302 proxy_b = ProxyFactory.create_proxy(calc, [LoggingAspect(), TimingAspect()], AllMethodsPointcut())
303 proxy_b.add(1, 1)
304
305 print("\nAll operations completed successfully.")Common Mistakes
- ✗Applying aspects to every method regardless of need, adding interception overhead to hot paths and hurting performance
- ✗Not preserving the method signature on the proxy, so callers get attribute errors or wrong return types
- ✗Aspect order dependency bugs: changing the order silently changes behavior, and nobody notices until production
- ✗Self-invocation bypassing the proxy: when a method calls this.otherMethod() internally, the call goes directly to the real object and skips all aspects
Key Points
- ✓Proxies intercept method calls without modifying the target class. The target has no idea it is being wrapped.
- ✓Pointcuts select which methods get advice. You might apply logging to every public method but caching only to read operations.
- ✓Around advice controls the entire call lifecycle: it can inspect arguments, decide whether to proceed, modify the return value, or catch exceptions.
- ✓Aspect ordering matters. If logging runs before timing, the log includes the timing overhead. The chain defines the execution order explicitly.