Skip to main content

​
Signature

def shutdown() -> None

​
Parameters

None.

​
Example

from rdk import init, shutdown

init(endpoint="...", api_key="...")

# Your application code...

# Clean shutdown
shutdown()

​
Behavior

When called, shutdown():
  1. Stops the background flush timer
  2. Flushes all pending traces
  3. Cleans up the RDK client instance
  4. Removes SDK instrumentation

​
Use Cases

FastAPI lifespan: 
from contextlib import asynccontextmanager
from fastapi import FastAPI
from rdk import init, shutdown

@asynccontextmanager
async def lifespan(app: FastAPI):
    init(endpoint="...", api_key="...")
    yield
    shutdown()

app = FastAPI(lifespan=lifespan)
Script completion: 
from rdk import init, observe, shutdown

init(endpoint="...", api_key="...")

@observe(name="main")
def main():
    # Your code
    ...

if __name__ == "__main__":
    try:
        main()
    finally:
        shutdown()
Signal handling: 
import signal
from rdk import init, shutdown

init(endpoint="...", api_key="...")

def signal_handler(signum, frame):
    shutdown()
    exit(0)

signal.signal(signal.SIGINT, signal_handler)
signal.signal(signal.SIGTERM, signal_handler)

​
Importance

Always call shutdown() before your application exits. Otherwise, pending traces may be lost.
Traces are batched in memory. Without shutdown():
  • Traces below the batch size won’t be sent
  • Traces waiting for the flush interval won’t be sent

​
Thread Safety

shutdown() is thread-safe but should only be called once.

​
See Also

  • init() - Initialize the client
  • flush() - Manual flush without shutdown