"""
Copyright 2026 OÜ KAVAL AI (registry code 17393877)
Licensed under the Apache License, Version 2.0 (the "License");
you may not use this file except in compliance with the License.
You may obtain a copy of the License at
http://www.apache.org/licenses/LICENSE-2.0
Unless required by applicable law or agreed to in writing, software
distributed under the License is distributed on an "AS IS" BASIS,
WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
See the License for the specific language governing permissions and
limitations under the License.
"""
from typing import List, Dict, Any, Optional
from uuid import UUID
from fastapi import HTTPException, status
from loguru import logger
from sqlalchemy import func, select, text, update as sa_update, delete as sa_delete
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker
from kavalai.backoffice import db
from kavalai.crud import insert, update, delete
from kavalai.db import build_db_uri, db_manager
PROJECT_FIELDS = frozenset(
{
"name",
"description",
"db_type",
"db_host",
"db_port",
"db_user",
"db_password",
"db_name",
"db_schema",
"rag_schema",
"read_only",
}
)
"""The fields of a project a client may set: its name, description and
connection settings. Everything else a project row carries — ``id``, the
timestamps, the caller's ``role`` the list endpoint adds — is the server's."""
[docs]
def project_fields(data: Dict[str, Any]) -> Dict[str, Any]:
"""The part of a request body that may become project columns.
The projects page sends the whole project back as it listed it, so a
body may carry ``created_at`` as an ISO string, which the database
refuses for a timestamp column; keys outside :data:`PROJECT_FIELDS` are
dropped rather than written.
"""
return {key: value for key, value in data.items() if key in PROJECT_FIELDS}
[docs]
class ProjectService:
def __init__(self, session_maker: async_sessionmaker[AsyncSession]):
self.session_maker = session_maker
[docs]
async def get_user_projects(self, user_id: UUID) -> List[Dict[str, Any]]:
async with self.session_maker() as session:
return await db.get_user_projects(session, user_id)
[docs]
async def create_project(self, data: Dict[str, Any], owner_id: UUID) -> db.Project:
async with self.session_maker() as session:
new_project = await insert(session, db.Project, project_fields(data))
# The creator is the project's first owner.
membership_data = {
"user_id": owner_id,
"project_id": new_project.id,
"role": db.ProjectRole.owner,
}
await insert(session, db.ProjectMembership, membership_data)
return new_project
[docs]
async def update_project(
self, project_id: UUID, data: Dict[str, Any]
) -> Optional[db.Project]:
async with self.session_maker() as session:
return await update(session, db.Project, project_id, project_fields(data))
[docs]
async def delete_project(self, project_id: UUID) -> bool:
async with self.session_maker() as session:
# Detach the project from anyone who has it selected first, so the
# delete cannot fail on the users.active_project_id foreign key and
# nobody is left pointing at a project that no longer exists.
await self._clear_active_project(session, project_id)
return await delete(session, db.Project, project_id)
@staticmethod
async def _clear_active_project(
session: AsyncSession, project_id: UUID, user_id: Optional[UUID] = None
) -> None:
"""Clear ``active_project_id`` for users pointing at ``project_id``."""
stmt = sa_update(db.User).where(db.User.active_project_id == project_id)
if user_id is not None:
stmt = stmt.where(db.User.id == user_id)
await session.execute(stmt.values(active_project_id=None))
await session.commit()
[docs]
async def get_members(self, project_id: UUID) -> List[Dict[str, Any]]:
stmt = (
select(db.User, db.ProjectMembership.role)
.join(db.ProjectMembership, db.User.id == db.ProjectMembership.user_id)
.where(db.ProjectMembership.project_id == project_id)
)
async with self.session_maker() as session:
result = await session.execute(stmt)
return [
{
"id": str(user.id),
"name": user.name,
"email": user.email,
"picture": user.picture,
"role": role.value,
}
for user, role in result.all()
]
[docs]
async def add_member(
self, project_id: UUID, user_id: UUID, role: db.ProjectRole
) -> None:
async with self.session_maker() as session:
if await db.is_member(session, user_id, project_id):
raise HTTPException(status_code=400, detail="User is already a member.")
membership_data = {
"user_id": user_id,
"project_id": project_id,
"role": role,
}
await insert(session, db.ProjectMembership, membership_data)
@staticmethod
async def _get_membership(
session: AsyncSession, project_id: UUID, user_id: UUID
) -> db.ProjectMembership:
stmt = select(db.ProjectMembership).where(
db.ProjectMembership.project_id == project_id,
db.ProjectMembership.user_id == user_id,
)
membership = (await session.execute(stmt)).scalars().first()
if not membership:
raise HTTPException(status_code=404, detail="Membership not found.")
return membership
@staticmethod
async def _assert_not_last_owner(
session: AsyncSession, membership: db.ProjectMembership, detail: str
) -> None:
"""Refuse to strip ownership from a project's only owner."""
if membership.role != db.ProjectRole.owner:
return
stmt = (
select(func.count())
.select_from(db.ProjectMembership)
.where(
db.ProjectMembership.project_id == membership.project_id,
db.ProjectMembership.role == db.ProjectRole.owner,
)
)
if (await session.execute(stmt)).scalar_one() <= 1:
raise HTTPException(status_code=400, detail=detail)
[docs]
async def update_member_role(
self, project_id: UUID, user_id: UUID, new_role: db.ProjectRole
) -> None:
async with self.session_maker() as session:
membership = await self._get_membership(session, project_id, user_id)
if new_role != db.ProjectRole.owner:
await self._assert_not_last_owner(
session, membership, "Cannot demote the last owner of the project."
)
stmt = (
sa_update(db.ProjectMembership)
.where(
db.ProjectMembership.project_id == project_id,
db.ProjectMembership.user_id == user_id,
)
.values(role=new_role)
)
await session.execute(stmt)
await session.commit()
[docs]
async def remove_member(self, project_id: UUID, user_id: UUID) -> None:
async with self.session_maker() as session:
membership = await self._get_membership(session, project_id, user_id)
await self._assert_not_last_owner(
session, membership, "Cannot remove the last owner of the project."
)
stmt = sa_delete(db.ProjectMembership).where(
db.ProjectMembership.project_id == project_id,
db.ProjectMembership.user_id == user_id,
)
await session.execute(stmt)
await session.commit()
# Without this the removed user keeps the project selected and
# every project-scoped request answers 403.
await self._clear_active_project(session, project_id, user_id)
[docs]
async def test_connection(self, project: db.Project) -> Dict[str, str]:
try:
logger.info(
f"Testing connection to project database: "
f"{describe_project_database(project)}"
)
project_session_maker = project_sessionmaker(project)
async with project_session_maker() as project_session:
await project_session.execute(text("SELECT 1"))
return {"status": "success", "message": "Connection successful"}
except Exception as e:
logger.error(f"Failed to connect to project database: {e}")
raise HTTPException(
status_code=status.HTTP_400_BAD_REQUEST,
detail=f"Failed to connect: {str(e)}",
)
[docs]
def project_db_uri(project: db.Project) -> str:
"""The URI of the agent database a project points at.
``db_type`` decides how the connection fields are read: ``sqlite`` takes
``db_name`` as a file path, anything else builds a PostgreSQL URI from
host, port, user, password and database.
"""
if project.db_type == "sqlite":
return f"sqlite:///{project.db_name}"
return build_db_uri(
project.db_user,
project.db_password,
project.db_host,
project.db_port,
project.db_name,
)
[docs]
def project_schema(project: db.Project) -> Optional[str]:
"""The schema of the project's agent database (``None`` for SQLite)."""
return None if project.db_type == "sqlite" else project.db_schema
[docs]
def project_rag_schema(project: db.Project) -> Optional[str]:
"""The schema of the project's RAG collections: ``rag_schema`` when set,
otherwise the agent database's own schema (``None`` for SQLite)."""
if project.db_type == "sqlite":
return None
return project.rag_schema or project_schema(project)
[docs]
def project_read_only(project: db.Project) -> bool:
"""Whether the backoffice may only read the project's agent database.
On unless the project says ``False``: a project built from a form
(``Project(**data)`` on the connection test) carries no column default,
and the safe reading of "unsaid" is read-only.
"""
return project.read_only is not False
[docs]
def project_sessionmaker(project: db.Project) -> async_sessionmaker[AsyncSession]:
"""A sessionmaker for the agent database the project points at.
A read-only project gets connections the database itself holds to reads
(:meth:`~kavalai.db.DatabaseManager.get_sessionmaker`).
"""
return db_manager.get_sessionmaker(
uri=project_db_uri(project),
schema=project_schema(project),
read_only=project_read_only(project),
)
[docs]
def describe_project_database(project: db.Project) -> str:
"""A log-safe description of the project's database (no password)."""
access = "read-only" if project_read_only(project) else "read-write"
if project.db_type == "sqlite":
return f"sqlite file={project.db_name}, {access}"
return (
f"host={project.db_host}, port={project.db_port}, db={project.db_name}, "
f"user={project.db_user}, schema={project.db_schema}, "
f"rag_schema={project_rag_schema(project)}, {access}"
)