AutoTrace add parameters and result

[danerlt]

Description

I have a simple flask todo list app. Its code is as follows, its name is flask_app.py:

import time

from flask import Flask
from flask_restx import Api, Resource, fields
from pydantic import BaseModel
from typing import List, Optional, Dict
from datetime import datetime

app = Flask(__name__)
api = Api(app, title="Todo List API", description="A simple Todo List API")


# Pydantic model
class TodoItem(BaseModel):
    id: int
    title: str
    description: Optional[str] = None
    completed: bool = False
    created_at: datetime
    updated_at: datetime


# Todo Service class
class TodoService:
    def __init__(self):
        self.todos: Dict[int, TodoItem] = {}
        self.counter = 1

    def create_todo(self, title: str, description: str = None) -> TodoItem:
        time.sleep(0.1)  # Simulate time-consuming operation
        todo = TodoItem(
            id=self.counter, title=title, description=description, created_at=datetime.now(), updated_at=datetime.now()
        )
        self.todos[self.counter] = todo
        self.counter += 1
        return todo

    def get_all_todos(self) -> List[TodoItem]:
        time.sleep(0.1)  # Simulate time-consuming operation
        return list(self.todos.values())

    def get_todo(self, todo_id: int) -> Optional[TodoItem]:
        return self.todos.get(todo_id)

    def update_todo(
        self, todo_id: int, title: str = None, description: str = None, completed: bool = None
    ) -> Optional[TodoItem]:
        if todo_id not in self.todos:
            return None

        time.sleep(1)  # Simulate time-consuming operation
        todo = self.todos[todo_id]
        update_data = {}

        if title is not None:
            update_data["title"] = title
        if description is not None:
            update_data["description"] = description
        if completed is not None:
            update_data["completed"] = completed

        update_data["updated_at"] = datetime.now()

        updated_todo = todo.copy(update=update_data)
        self.todos[todo_id] = updated_todo
        return updated_todo

    def delete_todo(self, todo_id: int) -> bool:
        if todo_id in self.todos:
            del self.todos[todo_id]
            return True
        return False


# Create service instance
todo_service = TodoService()

# API model definition
todo_model = api.model(
    "Todo",
    {
        "title": fields.String(required=True, description="Todo title"),
        "description": fields.String(description="Todo description"),
        "completed": fields.Boolean(description="Completion status"),
    },
)

todo_response_model = api.model(
    "TodoResponse",
    {
        "id": fields.Integer(description="Todo ID"),
        "title": fields.String(description="Todo title"),
        "description": fields.String(description="Todo description"),
        "completed": fields.Boolean(description="Completion status"),
        "created_at": fields.DateTime(description="Creation time"),
        "updated_at": fields.DateTime(description="Update time"),
    },
)


# API routes
@api.route("/todos")
class TodoList(Resource):
    @api.marshal_list_with(todo_response_model)
    def get(self):
        """Get all Todo items"""
        return [todo.dict() for todo in todo_service.get_all_todos()]

    @api.expect(todo_model)
    @api.marshal_with(todo_response_model)
    def post(self):
        """Create a new Todo item"""
        data = api.payload
        todo = todo_service.create_todo(data["title"], data.get("description"))
        return todo.dict()


@api.route("/todos/<int:todo_id>")
class Todo(Resource):
    @api.marshal_with(todo_response_model)
    def get(self, todo_id):
        """Get a specific Todo item"""
        todo = todo_service.get_todo(todo_id)
        if todo is None:
            api.abort(404, f"Todo {todo_id} not found")
        return todo.dict()

    @api.expect(todo_model)
    @api.marshal_with(todo_response_model)
    def put(self, todo_id):
        """Update a specific Todo item"""
        data = api.payload
        todo = todo_service.update_todo(
            todo_id, title=data.get("title"), description=data.get("description"), completed=data.get("completed")
        )
        if todo is None:
            api.abort(404, f"Todo {todo_id} not found")
        return todo.dict()

    def delete(self, todo_id):
        """Delete a specific Todo item"""
        if todo_service.delete_todo(todo_id):
            return {"message": f"Todo {todo_id} has been deleted"}, 200
        api.abort(404, f"Todo {todo_id} not found")


def main():
    app.run(debug=True)


if __name__ == "__main__":
    main()

Then I used the autotrace feature of logfire. The reference link is: https://logfire.pydantic.dev/docs/guides/onboarding-checklist/add-auto-tracing/.

The content of the code main.py that I started is as follows:

import logfire
from dotenv import load_dotenv

load_dotenv()
logfire.configure()
logfire.install_auto_tracing(modules=["flask_app"], min_duration=0.01)

from flask_app import main

main()

Then when I was browsing the results on the browser, I found that autotrace did not record the parameters and return results of the methods in the service. I hope to record the parameters and return values of this method as well.

[danerlt]

When I modify my main.py and integrate Flask into it, I still cannot see the parameters and return values.

new main.py

import logfire
from dotenv import load_dotenv

load_dotenv()
logfire.configure()
logfire.install_auto_tracing(modules=["flask_app"], min_duration=0.01)

from flask_app import app, main

logfire.instrument_flask(app=app)

main()

I still cannot see the parameters and return values !

[Kludex]

We need to work on this feature. It would be opt-in, since it's a lot of data.