### Install Autodynatrace via Pip

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Install the autodynatrace library using pip. This is the first step to enable automatic instrumentation.

```bash
pip install autodynatrace
```

--------------------------------

### Configure Autodynatrace via Environment Variables

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Demonstrates how to configure OneAgent SDK Python autoinstrumentation using environment variables for various settings like bootstrap, log level, fork support, header capture, and disabling specific instrumentation.

```bash
# Enable automatic instrumentation without code changes
export AUTOWRAPT_BOOTSTRAP=autodynatrace

# Set log level (DEBUG, INFO, WARNING, ERROR)
export AUTODYNATRACE_LOG_LEVEL=DEBUG

# Enable fork support for gunicorn/uwsgi
export AUTODYNATRACE_FORKABLE=True

# Capture HTTP request headers in traces
export AUTODYNATRACE_CAPTURE_HEADERS=True

# Override default web application info
export AUTODYNATRACE_VIRTUAL_HOST=api.myapp.com
export AUTODYNATRACE_APPLICATION_ID="My Application"
export AUTODYNATRACE_CONTEXT_ROOT=/api/v1

# Custom service name for @trace decorator
export AUTODYNATRACE_CUSTOM_SERVICE_NAME=MyCustomService

# Use fully qualified names (module.class.method)
export AUTODYNATRACE_CUSTOM_SERVICE_USE_FQN=True

# Disable specific library instrumentation
export AUTODYNATRACE_INSTRUMENT_REDIS=False
export AUTODYNATRACE_INSTRUMENT_CELERY=False
export AUTODYNATRACE_INSTRUMENT_CONCURRENT=False

# Enable async instrumentation (disabled by default)
export AUTODYNATRACE_INSTRUMENT_ASYNCIO=True

# Run application
python app.py
```

--------------------------------

### Enable Semi-Auto Instrumentation via Import

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Import the autodynatrace library at the beginning of your application to automatically instrument all supported libraries upon import. This approach requires a single import statement.

```python
import autodynatrace

# Your application code follows - all supported libraries are automatically instrumented
from flask import Flask
import redis
import pymongo

app = Flask(__name__)

@app.route('/')
def hello():
    # Redis and MongoDB calls will be automatically traced
    r = redis.Redis()
    r.get('key')
    return 'Hello World'

if __name__ == '__main__':
    app.run()
```

--------------------------------

### Django Application Instrumentation Configuration

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Configure Django to use autodynatrace middleware by adding 'autodynatrace.wrappers.django' to your INSTALLED_APPS in settings.py. Alternatively, import autodynatrace in wsgi.py or manage.py for auto-instrumentation.

```python
# settings.py
INSTALLED_APPS = [
    'django.contrib.admin',
    'django.contrib.auth',
    'django.contrib.contenttypes',
    'django.contrib.sessions',
    'django.contrib.messages',
    'django.contrib.staticfiles',
    # Add autodynatrace for automatic instrumentation
    'autodynatrace.wrappers.django',
]

# Alternative: auto-instrumentation via import in wsgi.py or manage.py
# import autodynatrace
```

--------------------------------

### Enable Auto-Instrumentation via Environment Variable

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Enable automatic instrumentation without modifying application code by setting the AUTOWRAPT_BOOTSTRAP environment variable to 'autodynatrace' before running your Python application.

```bash
# Set environment variable before running your application
export AUTOWRAPT_BOOTSTRAP=autodynatrace

# Run your Python application
python app.py
```

--------------------------------

### Complete Flask Application with AutoDynatrace

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

This snippet shows a full Flask application integrating multiple instrumented technologies. Ensure AUTODYNATRACE_LOG_LEVEL and AUTODYNATRACE_CAPTURE_HEADERS are set for debugging and header capture.

```python
import os
os.environ['AUTODYNATRACE_LOG_LEVEL'] = 'DEBUG'
os.environ['AUTODYNATRACE_CAPTURE_HEADERS'] = 'True'

import autodynatrace
from flask import Flask, jsonify, request
import redis
import psycopg2
from celery import Celery

app = Flask(__name__)

# Redis client
redis_client = redis.Redis(host='localhost', port=6379)

# Celery app
celery_app = Celery('tasks', broker='redis://localhost:6379/0')

@celery_app.task
def send_welcome_email(user_id):
    # Async task traced on worker
    return f"Welcome email sent to user {user_id}"

@autodynatrace.trace("UserService")
def get_user_from_db(user_id):
    # Custom traced function
    conn = psycopg2.connect(
        host='localhost', dbname='myapp',
        user='postgres', password='secret'
    )
    cursor = conn.cursor()
    cursor.execute("SELECT * FROM users WHERE id = %s", (user_id,))
    user = cursor.fetchone()
    conn.close()
    return user

@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    # Check cache first (Redis traced)
    cached = redis_client.get(f'user:{user_id}')
    if cached:
        return jsonify({'source': 'cache', 'data': cached.decode()})

    # Query database (psycopg2 traced)
    user = get_user_from_db(user_id)
    if user:
        # Cache result (Redis traced)
        redis_client.setex(f'user:{user_id}', 3600, str(user))
        return jsonify({'source': 'db', 'data': user})

    return jsonify({'error': 'User not found'}), 404

@app.route('/api/users', methods=['POST'])
def create_user():
    data = request.json
    conn = psycopg2.connect(
        host='localhost', dbname='myapp',
        user='postgres', password='secret'
    )
    cursor = conn.cursor()
    cursor.execute(
        "INSERT INTO users (name, email) VALUES (%s, %s) RETURNING id",
        (data['name'], data['email'])
    )
    user_id = cursor.fetchone()[0]
    conn.commit()
    conn.close()

    # Trigger async task (Celery traced)
    send_welcome_email.delay(user_id)

    return jsonify({'id': user_id}), 201

if __name__ == '__main__':
    app.run(debug=True)

```

--------------------------------

### Instrument PostgreSQL Queries

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

PostgreSQL queries executed via psycopg2 are traced with query text and row counts.

```python
import autodynatrace
import psycopg2

# Connection automatically instrumented with host, port, database info
conn = psycopg2.connect(
    host='localhost',
    port=5432,
    dbname='myapp',
    user='postgres',
    password='secret'
)

# Cursor operations traced with query text
cursor = conn.cursor()

# SELECT query - traces query and row count
cursor.execute("SELECT id, name, email FROM users WHERE active = %s", (True,))
users = cursor.fetchall()

# INSERT query traced
cursor.execute(
    "INSERT INTO orders (user_id, total, status) VALUES (%s, %s, %s) RETURNING id",
    (1, 99.99, 'pending')
)
order_id = cursor.fetchone()[0]

# UPDATE query traced
cursor.execute(
    "UPDATE orders SET status = %s WHERE id = %s",
    ('completed', order_id)
)

# Commit and close
conn.commit()
cursor.close()
conn.close()
```

--------------------------------

### Instrument SQLAlchemy ORM operations

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Automatically traces session management, database queries, and commits. Requires the autodynatrace package to be imported.

```python
import autodynatrace
from sqlalchemy import create_engine, Column, Integer, String
from sqlalchemy.orm import sessionmaker, declarative_base

Base = declarative_base()

class User(Base):
    __tablename__ = 'users'
    id = Column(Integer, primary_key=True)
    name = Column(String(50))
    email = Column(String(100))

# Create engine - database info captured for tracing
engine = create_engine('postgresql://postgres:secret@localhost:5432/myapp')
Session = sessionmaker(bind=engine)

# Session operations traced as custom services
session = Session()  # Session.init traced

# Query execution traced with SQL statement
users = session.query(User).filter(User.name.like('%John%')).all()

# Insert operation traced
new_user = User(name='Jane Doe', email='jane@example.com')
session.add(new_user)
session.commit()  # Commits traced as database request

# Context manager usage
with Session() as session:
    user = session.query(User).filter_by(id=1).first()
    if user:
        user.email = 'updated@example.com'
        session.commit()

session.close()  # Session.close traced
```

--------------------------------

### Instrument FastAPI Applications

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

FastAPI endpoints and response serialization are automatically instrumented.

```python
import autodynatrace
from fastapi import FastAPI
from pydantic import BaseModel

app = FastAPI()

class Item(BaseModel):
    name: str
    price: float

@app.get("/items/{item_id}")
async def read_item(item_id: int):
    # Endpoint automatically traced with function name "read_item"
    return {"item_id": item_id, "name": "Sample Item"}

@app.post("/items/")
async def create_item(item: Item):
    # POST endpoints also traced, including response serialization
    return {"name": item.name, "price": item.price}

@app.get("/users/{user_id}")
async def get_user(user_id: int, include_orders: bool = False):
    # Query parameters captured in trace context
    user = {"id": user_id, "name": "John"}
    if include_orders:
        user["orders"] = []
    return user

# Run with: uvicorn app:app --host 0.0.0.0 --port 8000
```

--------------------------------

### Instrument MongoDB operations with PyMongo

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Traces MongoDB operations including inserts, finds, updates, and aggregations via the command monitoring interface.

```python
import autodynatrace
from pymongo import MongoClient

# Connect to MongoDB - operations traced with host and port
client = MongoClient('mongodb://localhost:27017/')
db = client['myapp']

# Insert operations traced with collection name and operation type
result = db.users.insert_one({
    'name': 'John Doe',
    'email': 'john@example.com',
    'age': 30
})
user_id = result.inserted_id

# Find operations traced
user = db.users.find_one({'_id': user_id})
active_users = list(db.users.find({'active': True}).limit(10))

# Update operations traced
db.users.update_one(
    {'_id': user_id},
    {'$set': {'last_login': '2024-01-15'}}
)

# Aggregation pipeline traced
pipeline = [
    {'$match': {'active': True}},
    {'$group': {'_id': '$department', 'count': {'$sum': 1}}}
]
results = list(db.users.aggregate(pipeline))

# Delete operations traced
db.users.delete_one({'_id': user_id})

client.close()
```

--------------------------------

### Instrument RabbitMQ messaging with Pika

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Enables distributed context propagation for message publishing and consumption. Dynatrace tags are automatically injected into and extracted from message headers.

```python
import autodynatrace
import pika

# Publisher - outgoing messages traced
connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost', port=5672)
)
channel = connection.channel()

channel.queue_declare(queue='orders')

# Publish message - Dynatrace tag automatically injected into message headers
channel.basic_publish(
    exchange='',
    routing_key='orders',
    body='{"order_id": "12345", "total": 99.99}',
    properties=pika.BasicProperties(
        delivery_mode=2,  # Persistent message
        content_type='application/json'
    )
)
print("Order message published")
connection.close()

# Consumer - incoming messages traced with tag extraction
def process_message(ch, method, properties, body):
    # Message processing automatically traced
    # Dynatrace tag extracted from headers for distributed tracing
    print(f"Processing: {body}")
    ch.basic_ack(delivery_tag=method.delivery_tag)

connection = pika.BlockingConnection(
    pika.ConnectionParameters(host='localhost', port=5672)
)
channel = connection.channel()
channel.basic_consume(queue='orders', on_message_callback=process_message)
channel.start_consuming()
```

--------------------------------

### Instrument Django Views

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Django views are automatically traced with URL, method, and status code information.

```python
from django.http import JsonResponse
from django.views import View

class UserView(View):
    def get(self, request, user_id):
        # Request automatically traced with URL, method, status code
        # View function traced as custom service "Django Views"
        return JsonResponse({"id": user_id, "name": "John Doe"})

class OrderView(View):
    def post(self, request):
        # All HTTP methods are traced
        return JsonResponse({"status": "created"}, status=201)
```

--------------------------------

### Custom Function Tracing with @trace Decorator

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Use the @autodynatrace.trace decorator to trace custom functions and methods. It automatically detects service and method names, or allows custom names to be specified.

```python
import autodynatrace

# Basic usage - service name derived from module, method name from function
@autodynatrace.trace
def process_order(order_id):
    # This function will appear in Dynatrace as service "mymodule" with method "process_order"
    return f"Processing order {order_id}"

# Specify custom service name
@autodynatrace.trace("OrderService")
def validate_order(order_data):
    # Appears as service "OrderService" with method "validate_order"
    return order_data.get('valid', False)

# Specify custom method name only
@autodynatrace.trace(method="custom_method")
def another_function():
    # Appears with default service name and method "custom_method"
    return True

# Specify both service and method names
@autodynatrace.trace("PaymentService", "process_payment")
def handle_payment(amount, currency):
    # Appears as service "PaymentService" with method "process_payment"
    return {"status": "success", "amount": amount, "currency": currency}

# Use on class methods
class OrderProcessor:
    @autodynatrace.trace
    def process(self, order):
        # Service name will be "OrderProcessor", method name will be "process"
        return self.validate(order)

    @autodynatrace.trace("OrderProcessor", "validate_order")
    def validate(self, order):
        return order is not None

# Execute traced functions
result = process_order("12345")
payment = handle_payment(99.99, "USD")
processor = OrderProcessor()
processor.process({"id": "12345"})
```

--------------------------------

### Instrument Redis Operations

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Redis commands and pipeline operations are automatically traced.

```python
import autodynatrace
import redis

# Connect to Redis - connection automatically instrumented
r = redis.Redis(host='localhost', port=6379, db=0)

# Basic operations traced as database requests
r.set('user:1000:name', 'John Doe')
name = r.get('user:1000:name')

# Hash operations
r.hset('user:1000', mapping={
    'name': 'John Doe',
    'email': 'john@example.com',
    'age': 30
})
user_data = r.hgetall('user:1000')

# List operations
r.rpush('queue:orders', 'order:1001', 'order:1002')
order = r.lpop('queue:orders')

# Pipeline operations traced as custom service
pipe = r.pipeline()
pipe.set('key1', 'value1')
pipe.set('key2', 'value2')
pipe.get('key1')
pipe.get('key2')
results = pipe.execute()  # All pipeline commands traced together

# Transactions
with r.pipeline(transaction=True) as pipe:
    pipe.incr('counter')
    pipe.expire('counter', 3600)
    pipe.execute()
```

--------------------------------

### Instrument Celery Tasks

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Celery tasks support distributed tracing between callers and workers.

```python
import autodynatrace
from celery import Celery

# Create Celery application
app = Celery('tasks', broker='redis://localhost:6379/0')

@app.task
def process_order(order_id):
    # Task execution automatically traced on the worker
    # Dynatrace tag propagated from caller to worker for distributed tracing
    return f"Processed order {order_id}"

@app.task
def send_notification(user_id, message):
    # All Celery tasks traced with routing key and queue information
    return f"Sent '{message}' to user {user_id}"

# Caller application - traces outgoing message
def create_order(order_data):
    order_id = save_order(order_data)
    # Async task call traced as outgoing message to Celery queue
    process_order.delay(order_id)
    send_notification.delay(order_data['user_id'], "Order received")
    return order_id

# Worker startup command:
# celery -A tasks worker --loglevel=info
# Note: Workers automatically reinitialize SDK with forkable=True
```

--------------------------------

### Instrument gRPC client calls

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Captures method, target, and status code information for gRPC client calls. Failures are automatically captured in the trace.

```python
import autodynatrace
import grpc
from myservice_pb2 import GetUserRequest
from myservice_pb2_grpc import UserServiceStub

# Create gRPC channel
channel = grpc.insecure_channel('localhost:50051')
stub = UserServiceStub(channel)

# Unary call - traced as outgoing web request with gRPC URL
request = GetUserRequest(user_id=12345)
response = stub.GetUser(request)
# Traced as: grpc://localhost:50051/UserService/GetUser

# Unary call with metadata
metadata = [('authorization', 'Bearer token123')]
response = stub.GetUser(request, metadata=metadata)

# Error handling - failures captured in trace
try:
    response = stub.GetUser(GetUserRequest(user_id=-1))
except grpc.RpcError as e:
    # Error status captured (e.g., NOT_FOUND maps to 500)
    print(f"gRPC error: {e.code()}")

channel.close()
```

--------------------------------

### Flask Application Instrumentation

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Flask applications are automatically instrumented when autodynatrace is imported. This captures incoming HTTP requests, including URLs, methods, status codes, and exceptions.

```python
import autodynatrace
from flask import Flask, jsonify

app = Flask(__name__)

@app.route('/api/users/<int:user_id>')
def get_user(user_id):
    # Request automatically traced with URL, method, and status code
    return jsonify({"id": user_id, "name": "John Doe"})

@app.route('/api/orders', methods=['POST'])
def create_order():
    # POST requests also automatically traced
    return jsonify({"status": "created"}), 201

@app.route('/api/error')
def trigger_error():
    # Exceptions are automatically captured and marked as failed in Dynatrace
    raise ValueError("Something went wrong")

if __name__ == '__main__':
    # Run with gunicorn for production using forkable mode
    # AUTODYNATRACE_FORKABLE=True gunicorn -w 4 app:app
    app.run(debug=True)
```

--------------------------------

### Instrument aiohttp Client Requests

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Automatically traces outgoing HTTP requests made with aiohttp. Ensures distributed tracing headers are added and response status codes are captured.

```python
import autodynatrace
import aiohttp
import asyncio

async def fetch_data():
    async with aiohttp.ClientSession() as session:
        # GET request traced as outgoing web request
        # X-Dynatrace header automatically added for distributed tracing
        async with session.get('https://api.example.com/users/1') as response:
            # Status code captured in trace
            user = await response.json()
            return user

async def post_data():
    async with aioiohttp.ClientSession() as session:
        # POST request traced with URL and method
        async with session.post(
            'https://api.example.com/orders',
            json={'item': 'widget', 'quantity': 5},
            headers={'Content-Type': 'application/json'}
        ) as response:
            result = await response.json()
            return result

async def parallel_requests():
    async with aiohttp.ClientSession() as session:
        # Multiple concurrent requests all traced
        tasks = [
            session.get('https://api.example.com/users/1'),
            session.get('https://api.example.com/users/2'),
            session.get('https://api.example.com/products/100')
        ]
        responses = await asyncio.gather(*tasks)
        return [await r.json() for r in responses]

# Run async functions
user = asyncio.run(fetch_data())
results = asyncio.run(parallel_requests())
```

--------------------------------

### Instrument Subprocess Calls

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Automatically traces subprocess calls, capturing command arguments. Supports various subprocess functions including run, call, check_output, and Popen. Shell commands can also be traced.

```python
import autodynatrace
import subprocess

# subprocess.run traced (Python 3+)
result = subprocess.run(
    ['ls', '-la', '/tmp'],
    capture_output=True,
    text=True
)
print(result.stdout)

# subprocess.call traced
return_code = subprocess.call(['echo', 'Hello World'])

# subprocess.check_output traced
output = subprocess.check_output(['hostname'])
print(output.decode())

# subprocess.Popen traced
process = subprocess.Popen(
    ['grep', 'error', '/var/log/app.log'],
    stdout=subprocess.PIPE,
    stderr=subprocess.PIPE
)
stdout, stderr = process.communicate()

# Shell commands traced (use with caution)
result = subprocess.run(
    'ps aux | head -5',
    shell=True,
    capture_output=True,
    text=True
)
```

--------------------------------

### Instrument Concurrent Futures

Source: https://context7.com/dynatrace-oss/oneagent-sdk-python-autoinstrumentation/llms.txt

Traces thread pool operations using concurrent.futures, enabling in-process link propagation. Class method calls are also traced with their class names.

```python
import os
# Enable concurrent futures instrumentation
os.environ['AUTODYNATRACE_INSTRUMENT_CONCURRENT'] = 'True'

import autodynatrace
from concurrent.futures import ThreadPoolExecutor, as_completed

def process_item(item_id):
    # Function execution traced on worker thread
    # In-process link connects parent context to thread execution
    return f"Processed item {item_id}"

class DataProcessor:
    def process(self, data):
        # Class method calls also traced with class name
        return data.upper()

# ThreadPoolExecutor traced
with ThreadPoolExecutor(max_workers=4) as executor:
    # Submit tasks - parent context captured
    futures = [executor.submit(process_item, i) for i in range(10)]

    # Results collected - each execution traced
    for future in as_completed(futures):
        result = future.result()
        print(result)

# Map operation traced
with ThreadPoolExecutor(max_workers=4) as executor:
    processor = DataProcessor()
    items = ['item1', 'item2', 'item3']
    results = list(executor.map(processor.process, items))
```

=== COMPLETE CONTENT === This response contains all available snippets from this library. No additional content exists. Do not make further requests.