refactor(project): clean up _to_dict delegation and remove unused import

- Remove unused `import io`
- Avoid double-encoding in save_extracted_text by caching encoded bytes
- Delegate _to_dict queries to get_ontology/get_latest_graph_external_id
- Add comment to save_ontology noting full versioning is planned for F2-3

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Ubuntu 2026-05-03 01:16:03 +00:00
parent 949921344b
commit 8a6d00b26e
1 changed files with 8 additions and 19 deletions

View File

@ -1,6 +1,5 @@
"""Project context management — persistent via SQLAlchemy + StorageService.""" """Project context management — persistent via SQLAlchemy + StorageService."""
import uuid import uuid
import io
from datetime import datetime, timezone from datetime import datetime, timezone
from typing import Dict, Any, List, Optional from typing import Dict, Any, List, Optional
from enum import Enum from enum import Enum
@ -122,7 +121,9 @@ class ProjectManager:
@classmethod @classmethod
def save_extracted_text(cls, project_id: str, text: str, storage) -> None: def save_extracted_text(cls, project_id: str, text: str, storage) -> None:
storage_path = f"projects/{project_id}/extracted_text.txt" storage_path = f"projects/{project_id}/extracted_text.txt"
storage.upload(storage_path, text.encode("utf-8"), "text/plain") encoded = text.encode("utf-8")
size = len(encoded)
storage.upload(storage_path, encoded, "text/plain")
with get_session() as db: with get_session() as db:
from sqlalchemy import select from sqlalchemy import select
@ -133,14 +134,14 @@ class ProjectManager:
existing = db.execute(stmt).scalar_one_or_none() existing = db.execute(stmt).scalar_one_or_none()
if existing: if existing:
existing.storage_path = storage_path existing.storage_path = storage_path
existing.size = len(text.encode("utf-8")) existing.size = size
else: else:
rec = ProjectFileModel( rec = ProjectFileModel(
id=str(uuid.uuid4()), id=str(uuid.uuid4()),
project_id=project_id, project_id=project_id,
original_name="extracted_text.txt", original_name="extracted_text.txt",
storage_path=storage_path, storage_path=storage_path,
size=len(text.encode("utf-8")), size=size,
mime_type="text/plain", mime_type="text/plain",
file_type="extracted_text", file_type="extracted_text",
) )
@ -156,6 +157,7 @@ class ProjectManager:
@classmethod @classmethod
def save_ontology(cls, project_id: str, entity_types: list, edge_types: list) -> str: def save_ontology(cls, project_id: str, entity_types: list, edge_types: list) -> str:
# Upsert: versioning complet planificat a F2-3
from .db_models import OntologyModel from .db_models import OntologyModel
from sqlalchemy import select from sqlalchemy import select
with get_session() as db: with get_session() as db:
@ -240,21 +242,8 @@ class ProjectManager:
@classmethod @classmethod
def _to_dict(cls, proj: "ProjectModel") -> Dict[str, Any]: def _to_dict(cls, proj: "ProjectModel") -> Dict[str, Any]:
from .db_models import GraphModel, OntologyModel ontology = cls.get_ontology(proj.id)
from sqlalchemy import select graph_external_id = cls.get_latest_graph_external_id(proj.id)
graph_external_id = None
ontology = None
with get_session() as db2:
graph_rec = db2.execute(
select(GraphModel).where(GraphModel.project_id == proj.id).order_by(GraphModel.created_at.desc())
).scalars().first()
ont_rec = db2.execute(
select(OntologyModel).where(OntologyModel.project_id == proj.id).order_by(OntologyModel.version.desc())
).scalars().first()
if graph_rec:
graph_external_id = graph_rec.external_id
if ont_rec:
ontology = {"entity_types": ont_rec.entity_types or [], "edge_types": ont_rec.edge_types or []}
return { return {
"id": proj.id, "id": proj.id,
"project_id": proj.id, "project_id": proj.id,