opentelemetry-instrumentation-tortoiseorm

import os import asyncio from opentelemetry import trace from opentelemetry.sdk.resources import Resource from opentelemetry.sdk.trace import TracerProvider from opentelemetry.sdk.trace.export import ConsoleSpanExporter, SimpleSpanProcessor from opentelemetry.instrumentation.tortoiseorm import TortoiseORMInstrumentor # 1. Setup OpenTelemetry TracerProvider and Exporter resource = Resource.create({"service.name": "tortoise-orm-app"}) provider = TracerProvider(resource=resource) # Using ConsoleSpanExporter for demonstration; replace with OTLPSpanExporter for real apps processor = SimpleSpanProcessor(ConsoleSpanExporter()) provider.add_span_processor(processor) trace.set_tracer_provider(provider) # 2. Instrument Tortoise ORM TortoiseORMInstrumentor().instrument() # 3. Tortoise ORM application code from tortoise import models, fields, Tortoise class User(models.Model): id = fields.IntField(pk=True) name = fields.CharField(max_length=255) class Meta: table = "users" def __str__(self): return self.name async def main(): # Connect to the database and generate schema await Tortoise.init(db_url="sqlite://:memory:", modules={"models": ["__main__"]}) await Tortoise.generate_schemas() # Perform some ORM operations within a custom span to see the full flow tracer = trace.get_tracer(__name__) with tracer.start_as_current_span("application-flow"): with tracer.start_as_current_span("create_user"): user = await User.create(name="Alice") print(f"Created user: {user.name}") with tracer.start_as_current_span("fetch_user"): fetched_user = await User.get(id=user.id) print(f"Fetched user: {fetched_user.name}") with tracer.start_as_current_span("update_user"): fetched_user.name = "Alicia" await fetched_user.save() print(f"Updated user to: {fetched_user.name}") with tracer.start_as_current_span("delete_user"): await fetched_user.delete() print(f"Deleted user: {fetched_user.name}") # Close connection await Tortoise.close_connections() if __name__ == "__main__": asyncio.run(main())