Hi, are there apis to delete traces and spans?
Hi, are there apis to delete traces and spans?
Hi, are there apis to delete traces and spans?
Yes, Phoenix provides APIs to delete traces and spans. You can use the delete_traces function to delete specific traces by their IDs. Here's an example: LBKSfrom phoenix.server.types import DbSessionFactoryLBKSasync def delete_traces(db: DbSessionFactory, *trace_ids: str) -> List[int]: if not trace_ids: return [] stmt = ( delete(models.Trace) .where(models.Trace.trace_id.in_(set(trace_ids))) .returning(models.Trace.id) ) async with db() as session: return list(await session.scalars(stmt))LBKSFor deleting projects, you can use the delete_projects function similarly.Would you like more details?Sources:
- phoenix/server/api/utils.py
馃挕聽Hint: Mention RunLLM in the thread for followups.
RunLLM how to get a
I'm here to help with questions specifically about Phoenix, the AI observability platform. Could you please provide more details or rephrase your question related to Phoenix? For example, are you looking for information on how to get a specific trace, span, or project ID in Phoenix?
馃挕聽Hint: Mention RunLLM in the thread for followups.
RunLLM how to get a DbSessionFactory
To get a DbSessionFactory in Phoenix, you can use the provided fixtures in the test setup. Here is an example of how to create a DbSessionFactory for both SQLite and PostgreSQL: 1. SQLite Engine:
from phoenix.db.engines import aio_sqlite_engine
from sqlalchemy.engine import make_urlLBKS async def get_sqlite_engine() -> AsyncEngine:
engine = aio_sqlite_engine(make_url("sqlite+aiosqlite://"), migrate=False, shared_cache=False)
async with engine.begin() as conn:
await conn.run_sync(models.Base.metadata.create_all)
return engine
```LBKSLBKS2. **PostgreSQL Engine**:
```LBKS from sqlalchemy.ext.asyncio import AsyncEngine
from phoenix.db.engines import aio_postgresql_engine
from sqlalchemy.engine import make_urlLBKS async def get_postgresql_engine(postgresql_url: str) -> AsyncEngine:
engine = aio_postgresql_engine(make_url(postgresql_url), migrate=False)
async with engine.begin() as conn:
await conn.run_sync(models.Base.metadata.create_all)
return engine
```LBKSLBKS3. **Creating `DbSessionFactory`**:
```LBKS from phoenix.server.types import DbSessionFactoryLBKS def create_db_session_factory(engine: AsyncEngine) -> DbSessionFactory:
return DbSessionFactory(lambda: _db_with_lock(engine), dialect="sqlite" if "sqlite" in str(engine.url) else "postgresql")LBKS def _db_with_lock(engine: AsyncEngine) -> DbSessionFactory:
lock, db = asyncio.Lock(), _db(engine)
@contextlib.asynccontextmanager
async def db_session():
async with lock:
async with db() as session:
yield session
return db_session
```LBKSLBKSWould you like more details?LBKSSources:
- [conftest.py](https://github.com/arize-ai/phoenix/blob/main/tests/conftest.py)
馃挕聽Hint: Mention RunLLM in the thread for followups.
we don鈥檛 have a dedicated REST api for deleting specific spans right now
in the UI there鈥檚 a deletion method based on time range
does that meet your needs?