Compare commits
| Author | SHA1 | Date | |
|---|---|---|---|
|
|
cf8d72f73e | ||
|
|
2e7976e382 | ||
|
|
2a0b8fcab2 |
No files matched your search
@@ -34,7 +34,10 @@
|
|||||||
"Bash(curl -s -b /tmp/fg_cookies.txt http://localhost:8000__TRACKED_VAR__)",
|
"Bash(curl -s -b /tmp/fg_cookies.txt http://localhost:8000__TRACKED_VAR__)",
|
||||||
"Bash(sqlite3 app.db \"PRAGMA foreign_keys=ON; SELECT scope_model_id FROM augmented_parts WHERE id=1862;\")",
|
"Bash(sqlite3 app.db \"PRAGMA foreign_keys=ON; SELECT scope_model_id FROM augmented_parts WHERE id=1862;\")",
|
||||||
"Bash(sqlite3 app.db \"SELECT id, name FROM scope_models WHERE name='Viper';\")",
|
"Bash(sqlite3 app.db \"SELECT id, name FROM scope_models WHERE name='Viper';\")",
|
||||||
"Bash(sqlite3 app.db \"SELECT COUNT\\(*\\) FROM augmented_parts WHERE scope_model_id=\\(SELECT id FROM scope_models WHERE name='Viper'\\);\")"
|
"Bash(sqlite3 app.db \"SELECT COUNT\\(*\\) FROM augmented_parts WHERE scope_model_id=\\(SELECT id FROM scope_models WHERE name='Viper'\\);\")",
|
||||||
|
"Bash(sqlite3 app.db \"PRAGMA wal_checkpoint\\(FULL\\);\")",
|
||||||
|
"Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://127.0.0.1:8000/login)",
|
||||||
|
"Bash(curl -s -o /dev/null -w \"%{http_code}\\\\n\" http://127.0.0.1:8000/admin/imported-parts)"
|
||||||
]
|
]
|
||||||
}
|
}
|
||||||
}
|
}
|
||||||
@@ -0,0 +1,44 @@
|
|||||||
|
"""add jobs table
|
||||||
|
|
||||||
|
Revision ID: 04425ce3d4d5
|
||||||
|
Revises: a3f7c19b2d40
|
||||||
|
Create Date: 2026-08-05 15:21:09.527442
|
||||||
|
"""
|
||||||
|
|
||||||
|
from typing import Sequence, Union
|
||||||
|
|
||||||
|
from alembic import op
|
||||||
|
import sqlalchemy as sa
|
||||||
|
|
||||||
|
# revision identifiers, used by Alembic.
|
||||||
|
revision: str = "04425ce3d4d5"
|
||||||
|
down_revision: Union[str, None] = "a3f7c19b2d40"
|
||||||
|
branch_labels: Union[str, Sequence[str], None] = None
|
||||||
|
depends_on: Union[str, Sequence[str], None] = None
|
||||||
|
|
||||||
|
|
||||||
|
def upgrade() -> None:
|
||||||
|
# ### commands auto generated by Alembic - please adjust! ###
|
||||||
|
op.create_table(
|
||||||
|
"jobs",
|
||||||
|
sa.Column("id", sa.String(length=32), nullable=False),
|
||||||
|
sa.Column("kind", sa.String(length=64), nullable=False),
|
||||||
|
sa.Column("status", sa.String(length=16), nullable=False),
|
||||||
|
sa.Column("message", sa.Text(), nullable=False),
|
||||||
|
sa.Column("result", sa.JSON(), nullable=True),
|
||||||
|
sa.Column(
|
||||||
|
"created_at",
|
||||||
|
sa.DateTime(timezone=True),
|
||||||
|
server_default=sa.text("(CURRENT_TIMESTAMP)"),
|
||||||
|
nullable=False,
|
||||||
|
),
|
||||||
|
sa.Column("finished_at", sa.DateTime(timezone=True), nullable=True),
|
||||||
|
sa.PrimaryKeyConstraint("id"),
|
||||||
|
)
|
||||||
|
# ### end Alembic commands ###
|
||||||
|
|
||||||
|
|
||||||
|
def downgrade() -> None:
|
||||||
|
# ### commands auto generated by Alembic - please adjust! ###
|
||||||
|
op.drop_table("jobs")
|
||||||
|
# ### end Alembic commands ###
|
||||||
@@ -14,6 +14,7 @@ from app.models.fit import Fit, FitLabel, FitNote
|
|||||||
from app.models.imported_part import ImportedPart
|
from app.models.imported_part import ImportedPart
|
||||||
from app.models.imported_part_config import ImportedPartConfig
|
from app.models.imported_part_config import ImportedPartConfig
|
||||||
from app.models.iso_thread import IsoThread
|
from app.models.iso_thread import IsoThread
|
||||||
|
from app.models.job import Job
|
||||||
from app.models.manufacturer import Manufacturer
|
from app.models.manufacturer import Manufacturer
|
||||||
from app.models.part_colour import PartColour
|
from app.models.part_colour import PartColour
|
||||||
from app.models.part_type import PartType
|
from app.models.part_type import PartType
|
||||||
@@ -40,6 +41,7 @@ __all__ = [
|
|||||||
"ImportedPart",
|
"ImportedPart",
|
||||||
"ImportedPartConfig",
|
"ImportedPartConfig",
|
||||||
"IsoThread",
|
"IsoThread",
|
||||||
|
"Job",
|
||||||
"Manufacturer",
|
"Manufacturer",
|
||||||
"PartColour",
|
"PartColour",
|
||||||
"PartType",
|
"PartType",
|
||||||
|
|||||||
@@ -0,0 +1,31 @@
|
|||||||
|
from datetime import datetime
|
||||||
|
|
||||||
|
from sqlalchemy import JSON, DateTime, String, Text, func
|
||||||
|
from sqlalchemy.orm import Mapped, mapped_column
|
||||||
|
|
||||||
|
from app.core.database import Base
|
||||||
|
|
||||||
|
|
||||||
|
class Job(Base):
|
||||||
|
"""Tracks a long-running admin operation (sync, import, fit recompute).
|
||||||
|
|
||||||
|
Persisted (rather than kept in a module-level dict) so that any worker
|
||||||
|
process/replica handling a status-poll request can see progress recorded
|
||||||
|
by whichever process is actually running the job.
|
||||||
|
"""
|
||||||
|
|
||||||
|
__tablename__ = "jobs"
|
||||||
|
|
||||||
|
id: Mapped[str] = mapped_column(String(32), primary_key=True)
|
||||||
|
kind: Mapped[str] = mapped_column(String(64), nullable=False)
|
||||||
|
status: Mapped[str] = mapped_column(
|
||||||
|
String(16), nullable=False, default="running"
|
||||||
|
) # running | done | error
|
||||||
|
message: Mapped[str] = mapped_column(Text, nullable=False, default="Working…")
|
||||||
|
result: Mapped[dict | None] = mapped_column(JSON, nullable=True, default=None)
|
||||||
|
created_at: Mapped[datetime] = mapped_column(
|
||||||
|
DateTime(timezone=True), server_default=func.now()
|
||||||
|
)
|
||||||
|
finished_at: Mapped[datetime | None] = mapped_column(
|
||||||
|
DateTime(timezone=True), nullable=True, default=None
|
||||||
|
)
|
||||||
@@ -16,12 +16,9 @@ from app.models.fit import Fit, FitNote
|
|||||||
from app.models.imported_part import ImportedPart
|
from app.models.imported_part import ImportedPart
|
||||||
from app.models.iso_thread import IsoThread
|
from app.models.iso_thread import IsoThread
|
||||||
from app.models.manufacturer import Manufacturer
|
from app.models.manufacturer import Manufacturer
|
||||||
from app.models.part_colour import PartColour
|
|
||||||
from app.models.part_type import PartType
|
|
||||||
from app.models.scope_info import ScopeInfo
|
from app.models.scope_info import ScopeInfo
|
||||||
from app.models.scope_model import ScopeModel
|
from app.models.scope_model import ScopeModel
|
||||||
from app.models.static_connection import StaticConnection
|
from app.models.static_connection import StaticConnection
|
||||||
from app.models.sub_part_type import SubPartType
|
|
||||||
from app.models.unc_thread import UncThread
|
from app.models.unc_thread import UncThread
|
||||||
|
|
||||||
# Children before the RESTRICT-protected reference tables they point to.
|
# Children before the RESTRICT-protected reference tables they point to.
|
||||||
|
|||||||
+74
-52
@@ -1,11 +1,12 @@
|
|||||||
"""Lightweight in-process tracking for long-running admin operations.
|
"""DB-backed tracking for long-running admin operations.
|
||||||
|
|
||||||
Jobs live in a module-level dict, so this only works correctly for a single
|
Jobs are persisted to the ``jobs`` table rather than kept in a module-level
|
||||||
OS process (the app's current deployment: one uvicorn worker, no multi-process
|
dict, so that a status-poll request landing on a *different* worker
|
||||||
or multi-replica setup — see ``entrypoint.sh`` / ``docker-compose.yml``). If
|
process/replica than the one running the job can still see its progress —
|
||||||
the app is ever scaled to multiple workers/replicas, this needs to move to
|
required once the app runs with more than one uvicorn worker or replica
|
||||||
shared storage (a DB table, Redis) so a poll request landing on a different
|
(a module-level dict is process-local and would report "not found" for a job
|
||||||
process can still see the job.
|
started elsewhere, even while that job is still running and writing to the
|
||||||
|
shared database).
|
||||||
"""
|
"""
|
||||||
|
|
||||||
from __future__ import annotations
|
from __future__ import annotations
|
||||||
@@ -13,62 +14,83 @@ from __future__ import annotations
|
|||||||
import asyncio
|
import asyncio
|
||||||
import uuid
|
import uuid
|
||||||
from collections.abc import Awaitable, Callable
|
from collections.abc import Awaitable, Callable
|
||||||
from dataclasses import dataclass, field
|
from datetime import UTC, datetime, timedelta
|
||||||
from datetime import UTC, datetime
|
|
||||||
|
|
||||||
_MAX_FINISHED_AGE_SECONDS = 3600
|
from sqlalchemy import delete
|
||||||
|
from sqlalchemy.ext.asyncio import AsyncSession
|
||||||
|
|
||||||
|
from app.core.database import AsyncSessionLocal
|
||||||
|
from app.models.job import Job
|
||||||
|
|
||||||
|
_MAX_FINISHED_AGE = timedelta(hours=1)
|
||||||
|
|
||||||
|
|
||||||
@dataclass
|
async def get(db: AsyncSession, job_id: str) -> Job | None:
|
||||||
class Job:
|
return await db.get(Job, job_id)
|
||||||
id: str
|
|
||||||
kind: str
|
|
||||||
status: str = "running" # running | done | error
|
|
||||||
message: str = "Working…"
|
|
||||||
result: object | None = None
|
|
||||||
created_at: datetime = field(default_factory=lambda: datetime.now(UTC))
|
|
||||||
finished_at: datetime | None = None
|
|
||||||
|
|
||||||
|
|
||||||
_jobs: dict[str, Job] = {}
|
async def start(kind: str, run: Callable[[str], Awaitable[None]]) -> Job:
|
||||||
|
"""Create a job row, launch ``run(job_id)`` in the background, and return
|
||||||
|
the job immediately.
|
||||||
|
|
||||||
|
``run`` is responsible for calling ``mark_done``/``mark_error`` to record
|
||||||
def get(job_id: str) -> Job | None:
|
the outcome, including on its own errors. Any exception that escapes
|
||||||
return _jobs.get(job_id)
|
anyway is caught here as a last resort so the job doesn't hang forever in
|
||||||
|
"running" state.
|
||||||
|
|
||||||
def _evict_stale() -> None:
|
|
||||||
now = datetime.now(UTC)
|
|
||||||
stale = [
|
|
||||||
job_id
|
|
||||||
for job_id, job in _jobs.items()
|
|
||||||
if job.finished_at
|
|
||||||
and (now - job.finished_at).total_seconds() > _MAX_FINISHED_AGE_SECONDS
|
|
||||||
]
|
|
||||||
for job_id in stale:
|
|
||||||
del _jobs[job_id]
|
|
||||||
|
|
||||||
|
|
||||||
def start(kind: str, run: Callable[[Job], Awaitable[None]]) -> Job:
|
|
||||||
"""Create a job, launch ``run(job)`` in the background, and return it immediately.
|
|
||||||
|
|
||||||
``run`` is responsible for updating ``job.status``/``job.message``/``job.result``
|
|
||||||
to reflect the outcome, including catching its own errors. Any exception that
|
|
||||||
escapes anyway is caught here as a last resort so the job doesn't hang forever
|
|
||||||
in "running" state.
|
|
||||||
"""
|
"""
|
||||||
_evict_stale()
|
async with AsyncSessionLocal() as db:
|
||||||
job = Job(id=uuid.uuid4().hex, kind=kind)
|
await _evict_stale(db)
|
||||||
_jobs[job.id] = job
|
job = Job(id=uuid.uuid4().hex, kind=kind, status="running", message="Working…")
|
||||||
|
db.add(job)
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
async def _runner() -> None:
|
async def _runner() -> None:
|
||||||
try:
|
try:
|
||||||
await run(job)
|
await run(job.id)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
job.status = "error"
|
await mark_error(job.id, f"{kind} failed: {exc}")
|
||||||
job.message = f"{kind} failed: {exc}"
|
|
||||||
finally:
|
|
||||||
job.finished_at = datetime.now(UTC)
|
|
||||||
|
|
||||||
asyncio.create_task(_runner())
|
asyncio.create_task(_runner())
|
||||||
return job
|
return job
|
||||||
|
|
||||||
|
|
||||||
|
async def mark_done(job_id: str, message: str, result: dict | None = None) -> None:
|
||||||
|
await _update(job_id, status="done", message=message, result=result)
|
||||||
|
|
||||||
|
|
||||||
|
async def mark_error(job_id: str, message: str) -> None:
|
||||||
|
await _update(job_id, status="error", message=message)
|
||||||
|
|
||||||
|
|
||||||
|
async def set_message(job_id: str, message: str) -> None:
|
||||||
|
await _update(job_id, message=message)
|
||||||
|
|
||||||
|
|
||||||
|
async def _update(
|
||||||
|
job_id: str,
|
||||||
|
*,
|
||||||
|
status: str | None = None,
|
||||||
|
message: str | None = None,
|
||||||
|
result: dict | None = None,
|
||||||
|
) -> None:
|
||||||
|
async with AsyncSessionLocal() as db:
|
||||||
|
job = await db.get(Job, job_id)
|
||||||
|
if job is None:
|
||||||
|
return
|
||||||
|
if status is not None:
|
||||||
|
job.status = status
|
||||||
|
if status in ("done", "error"):
|
||||||
|
job.finished_at = datetime.now(UTC)
|
||||||
|
if message is not None:
|
||||||
|
job.message = message
|
||||||
|
if result is not None:
|
||||||
|
job.result = result
|
||||||
|
await db.commit()
|
||||||
|
|
||||||
|
|
||||||
|
async def _evict_stale(db: AsyncSession) -> None:
|
||||||
|
cutoff = datetime.now(UTC) - _MAX_FINISHED_AGE
|
||||||
|
await db.execute(
|
||||||
|
delete(Job).where(Job.finished_at.is_not(None), Job.finished_at < cutoff)
|
||||||
|
)
|
||||||
|
await db.commit()
|
||||||
@@ -8,15 +8,15 @@
|
|||||||
<div class="card" style="border: 1px solid var(--color-danger, #c00)">
|
<div class="card" style="border: 1px solid var(--color-danger, #c00)">
|
||||||
<h2>Purge database</h2>
|
<h2>Purge database</h2>
|
||||||
<p>
|
<p>
|
||||||
Permanently deletes every part, taxonomy entry, attachment point, fit,
|
Permanently deletes every part, attachment point, fit, and imported
|
||||||
and imported part in the database — manufacturers, part colours, part
|
part in the database — manufacturers, scope models, ISO/UNC threads,
|
||||||
types, sub part types, scope models, ISO/UNC threads, clamp/static
|
clamp/static connections, attachment point names, augmented parts,
|
||||||
connections, attachment point names, augmented parts, scope info,
|
scope info, attachment points, fits, fit notes, and imported parts.
|
||||||
attachment points, fits, fit notes, and imported parts.
|
|
||||||
</p>
|
</p>
|
||||||
<p>
|
<p>
|
||||||
<strong>Not</strong> deleted: users, roles & permissions, refresh
|
<strong>Not</strong> deleted: users, roles & permissions, refresh
|
||||||
tokens, the audit log, and the OData connection configuration.
|
tokens, the audit log, the OData connection configuration, and part
|
||||||
|
taxonomy (part colours, part types, sub part types).
|
||||||
</p>
|
</p>
|
||||||
<p class="muted">This cannot be undone. Take a database backup first if you might need this data again.</p>
|
<p class="muted">This cannot be undone. Take a database backup first if you might need this data again.</p>
|
||||||
<button class="btn btn-danger"
|
<button class="btn btn-danger"
|
||||||
|
|||||||
@@ -29,11 +29,9 @@
|
|||||||
<button class="btn btn-primary"
|
<button class="btn btn-primary"
|
||||||
hx-post="/admin/imported-parts/refresh"
|
hx-post="/admin/imported-parts/refresh"
|
||||||
hx-target="#imported-parts-section"
|
hx-target="#imported-parts-section"
|
||||||
hx-swap="outerHTML"
|
hx-swap="outerHTML">
|
||||||
hx-indicator="#refresh-indicator">
|
|
||||||
Refresh from OData
|
Refresh from OData
|
||||||
</button>
|
</button>
|
||||||
<span id="refresh-indicator" class="htmx-indicator muted"> Loading…</span>
|
|
||||||
</div>
|
</div>
|
||||||
|
|
||||||
{% include "admin/_imported_parts_table.html" %}
|
{% include "admin/_imported_parts_table.html" %}
|
||||||
|
|||||||
+118
-44
@@ -1366,6 +1366,19 @@ async def save_imported_part_config(
|
|||||||
return HTMLResponse("Configuration saved.", status_code=200)
|
return HTMLResponse("Configuration saved.", status_code=200)
|
||||||
|
|
||||||
|
|
||||||
|
_IMPORTED_PARTS_REFRESH_TARGET_ID = "imported-parts-section"
|
||||||
|
|
||||||
|
|
||||||
|
async def _imported_parts_table_ctx(
|
||||||
|
db: AsyncSession, *, sync_result: str | None = None, sync_error: str | None = None
|
||||||
|
) -> dict:
|
||||||
|
return {
|
||||||
|
**await _default_imported_parts_table_ctx(db),
|
||||||
|
"sync_result": sync_result,
|
||||||
|
"sync_error": sync_error,
|
||||||
|
}
|
||||||
|
|
||||||
|
|
||||||
@router.post("/imported-parts/refresh", response_class=HTMLResponse)
|
@router.post("/imported-parts/refresh", response_class=HTMLResponse)
|
||||||
async def refresh_imported_parts(
|
async def refresh_imported_parts(
|
||||||
request: Request,
|
request: Request,
|
||||||
@@ -1373,45 +1386,94 @@ async def refresh_imported_parts(
|
|||||||
actor: User = Depends(web_require("imported_parts:write")),
|
actor: User = Depends(web_require("imported_parts:write")),
|
||||||
):
|
):
|
||||||
config = await imported_part_config_crud.get(db)
|
config = await imported_part_config_crud.get(db)
|
||||||
|
|
||||||
async def _imported_table_ctx(sync_result=None, sync_error=None):
|
|
||||||
return {
|
|
||||||
**await _default_imported_parts_table_ctx(db),
|
|
||||||
"sync_result": sync_result,
|
|
||||||
"sync_error": sync_error,
|
|
||||||
}
|
|
||||||
|
|
||||||
if not config or not config.odata_url:
|
if not config or not config.odata_url:
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
request,
|
request,
|
||||||
"admin/_imported_parts_table.html",
|
"admin/_imported_parts_table.html",
|
||||||
await _imported_table_ctx(sync_error="OData URL is not configured."),
|
await _imported_parts_table_ctx(
|
||||||
|
db, sync_error="OData URL is not configured."
|
||||||
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
actor_id = actor.id
|
||||||
|
|
||||||
|
async def _run(job_id: str) -> None:
|
||||||
|
async with AsyncSessionLocal() as db:
|
||||||
|
config = await imported_part_config_crud.get(db)
|
||||||
|
if not config or not config.odata_url:
|
||||||
|
await jobs.mark_error(job_id, "OData URL is not configured.")
|
||||||
|
return
|
||||||
try:
|
try:
|
||||||
records = await odata_service.fetch(
|
records = await odata_service.fetch(
|
||||||
config.odata_url, config.username, config.password
|
config.odata_url, config.username, config.password
|
||||||
)
|
)
|
||||||
created, updated = await imported_part_crud.upsert_bulk(db, records)
|
created, updated = await imported_part_crud.upsert_bulk(db, records)
|
||||||
|
except Exception as exc:
|
||||||
|
await jobs.mark_error(job_id, f"Sync failed: {exc}")
|
||||||
|
return
|
||||||
|
actor_row = await user_crud.get(db, actor_id)
|
||||||
await audit_crud.log(
|
await audit_crud.log(
|
||||||
db,
|
db,
|
||||||
actor=actor,
|
actor=actor_row,
|
||||||
action="update",
|
action="update",
|
||||||
resource_type="ImportedPart",
|
resource_type="ImportedPart",
|
||||||
resource_id=None,
|
resource_id=None,
|
||||||
description=f"OData sync: {created} created, {updated} updated ({len(records)} total records)",
|
description=(
|
||||||
)
|
f"OData sync: {created} created, {updated} updated "
|
||||||
return templates.TemplateResponse(
|
f"({len(records)} total records)"
|
||||||
request,
|
|
||||||
"admin/_imported_parts_table.html",
|
|
||||||
await _imported_table_ctx(
|
|
||||||
sync_result=f"Sync complete: {created} created, {updated} updated."
|
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
except Exception as exc:
|
await jobs.mark_done(
|
||||||
|
job_id,
|
||||||
|
message=f"Sync complete: {created} created, {updated} updated.",
|
||||||
|
)
|
||||||
|
|
||||||
|
job = await jobs.start("imported_parts_refresh", _run)
|
||||||
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
|
"admin/_job_progress.html",
|
||||||
|
{
|
||||||
|
"container_id": _IMPORTED_PARTS_REFRESH_TARGET_ID,
|
||||||
|
"poll_url": f"/admin/imported-parts/refresh/jobs/{job.id}",
|
||||||
|
"job": job,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
|
||||||
|
|
||||||
|
@router.get("/imported-parts/refresh/jobs/{job_id}", response_class=HTMLResponse)
|
||||||
|
async def imported_parts_refresh_job_status(
|
||||||
|
request: Request,
|
||||||
|
job_id: str,
|
||||||
|
db: AsyncSession = Depends(get_db),
|
||||||
|
_: User = Depends(web_require("imported_parts:write")),
|
||||||
|
):
|
||||||
|
job = await jobs.get(db, job_id)
|
||||||
|
if job is None:
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
request,
|
request,
|
||||||
"admin/_imported_parts_table.html",
|
"admin/_imported_parts_table.html",
|
||||||
await _imported_table_ctx(sync_error=f"Sync failed: {exc}"),
|
await _imported_parts_table_ctx(
|
||||||
|
db, sync_error="Refresh job not found — it may have expired."
|
||||||
|
),
|
||||||
|
)
|
||||||
|
if job.status == "running":
|
||||||
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
|
"admin/_job_progress.html",
|
||||||
|
{
|
||||||
|
"container_id": _IMPORTED_PARTS_REFRESH_TARGET_ID,
|
||||||
|
"poll_url": f"/admin/imported-parts/refresh/jobs/{job_id}",
|
||||||
|
"job": job,
|
||||||
|
},
|
||||||
|
)
|
||||||
|
return templates.TemplateResponse(
|
||||||
|
request,
|
||||||
|
"admin/_imported_parts_table.html",
|
||||||
|
await _imported_parts_table_ctx(
|
||||||
|
db,
|
||||||
|
sync_result=job.message if job.status == "done" else None,
|
||||||
|
sync_error=job.message if job.status == "error" else None,
|
||||||
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
|
|
||||||
@@ -1511,13 +1573,12 @@ async def upload_scope_import(
|
|||||||
data = await file.read()
|
data = await file.read()
|
||||||
actor_id = actor.id
|
actor_id = actor.id
|
||||||
|
|
||||||
async def _run(job: jobs.Job) -> None:
|
async def _run(job_id: str) -> None:
|
||||||
async with AsyncSessionLocal() as db:
|
async with AsyncSessionLocal() as db:
|
||||||
try:
|
try:
|
||||||
result = await scope_import_service.import_from_bytes(db, data)
|
result = await scope_import_service.import_from_bytes(db, data)
|
||||||
except Exception as exc: # noqa: BLE001
|
except Exception as exc: # noqa: BLE001
|
||||||
job.status = "error"
|
await jobs.mark_error(job_id, f"Could not read spreadsheet: {exc}")
|
||||||
job.message = f"Could not read spreadsheet: {exc}"
|
|
||||||
return
|
return
|
||||||
actor_row = await user_crud.get(db, actor_id)
|
actor_row = await user_crud.get(db, actor_id)
|
||||||
await audit_crud.log(
|
await audit_crud.log(
|
||||||
@@ -1532,10 +1593,16 @@ async def upload_scope_import(
|
|||||||
f"{result.errors} errors"
|
f"{result.errors} errors"
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
job.status = "done"
|
await jobs.mark_done(
|
||||||
job.result = result
|
job_id,
|
||||||
|
message=(
|
||||||
|
f"Scope import complete: {result.created} created, "
|
||||||
|
f"{result.updated} updated."
|
||||||
|
),
|
||||||
|
result=result.model_dump(mode="json"),
|
||||||
|
)
|
||||||
|
|
||||||
job = jobs.start("scope_import", _run)
|
job = await jobs.start("scope_import", _run)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
request, "admin/_job_progress.html", _scope_import_progress_ctx(job)
|
request, "admin/_job_progress.html", _scope_import_progress_ctx(job)
|
||||||
)
|
)
|
||||||
@@ -1545,9 +1612,10 @@ async def upload_scope_import(
|
|||||||
async def scope_import_job_status(
|
async def scope_import_job_status(
|
||||||
request: Request,
|
request: Request,
|
||||||
job_id: str,
|
job_id: str,
|
||||||
|
db: AsyncSession = Depends(get_db),
|
||||||
_: User = Depends(web_require("augmented_parts:write")),
|
_: User = Depends(web_require("augmented_parts:write")),
|
||||||
):
|
):
|
||||||
job = jobs.get(job_id)
|
job = await jobs.get(db, job_id)
|
||||||
if job is None:
|
if job is None:
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
request,
|
request,
|
||||||
@@ -1718,13 +1786,12 @@ async def sync_augmented_parts(
|
|||||||
):
|
):
|
||||||
actor_id = actor.id
|
actor_id = actor.id
|
||||||
|
|
||||||
async def _run(job: jobs.Job) -> None:
|
async def _run(job_id: str) -> None:
|
||||||
async with AsyncSessionLocal() as db:
|
async with AsyncSessionLocal() as db:
|
||||||
try:
|
try:
|
||||||
result = await augmented_part_sync_service.sync(db)
|
result = await augmented_part_sync_service.sync(db)
|
||||||
except Exception as exc:
|
except Exception as exc:
|
||||||
job.status = "error"
|
await jobs.mark_error(job_id, f"Sync failed: {exc}")
|
||||||
job.message = f"Sync failed: {exc}"
|
|
||||||
return
|
return
|
||||||
actor_row = await user_crud.get(db, actor_id)
|
actor_row = await user_crud.get(db, actor_id)
|
||||||
await audit_crud.log(
|
await audit_crud.log(
|
||||||
@@ -1738,18 +1805,21 @@ async def sync_augmented_parts(
|
|||||||
f"{result.skipped} skipped (no type match)"
|
f"{result.skipped} skipped (no type match)"
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
job.status = "done"
|
await jobs.mark_done(
|
||||||
job.message = (
|
job_id,
|
||||||
f"Sync complete: {result.created} created, {result.deleted} deleted"
|
message=(
|
||||||
|
f"Sync complete: {result.created} created, "
|
||||||
|
f"{result.deleted} deleted"
|
||||||
+ (
|
+ (
|
||||||
f", {result.skipped} skipped (type not recognised)"
|
f", {result.skipped} skipped (type not recognised)"
|
||||||
if result.skipped
|
if result.skipped
|
||||||
else ""
|
else ""
|
||||||
)
|
)
|
||||||
+ "."
|
+ "."
|
||||||
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
job = jobs.start("augmented_part_sync", _run)
|
job = await jobs.start("augmented_part_sync", _run)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
request,
|
request,
|
||||||
"admin/_job_progress.html",
|
"admin/_job_progress.html",
|
||||||
@@ -1768,7 +1838,7 @@ async def augmented_parts_sync_job_status(
|
|||||||
db: AsyncSession = Depends(get_db),
|
db: AsyncSession = Depends(get_db),
|
||||||
_: User = Depends(web_require("augmented_parts:write")),
|
_: User = Depends(web_require("augmented_parts:write")),
|
||||||
):
|
):
|
||||||
job = jobs.get(job_id)
|
job = await jobs.get(db, job_id)
|
||||||
if job is None:
|
if job is None:
|
||||||
form_ctx = await _augmented_part_form_ctx(db)
|
form_ctx = await _augmented_part_form_ctx(db)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
@@ -2483,7 +2553,7 @@ async def recompute_augmented_part_fits(
|
|||||||
):
|
):
|
||||||
actor_id = actor.id
|
actor_id = actor.id
|
||||||
|
|
||||||
async def _run(job: jobs.Job) -> None:
|
async def _run(job_id: str) -> None:
|
||||||
async with AsyncSessionLocal() as db:
|
async with AsyncSessionLocal() as db:
|
||||||
result = await fit_matching_service.recompute_for_part(
|
result = await fit_matching_service.recompute_for_part(
|
||||||
db, augmented_part_id
|
db, augmented_part_id
|
||||||
@@ -2500,10 +2570,11 @@ async def recompute_augmented_part_fits(
|
|||||||
f"{result.created} created, {result.removed} removed"
|
f"{result.created} created, {result.removed} removed"
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
job.status = "done"
|
await jobs.mark_done(
|
||||||
job.message = f"{result.created} created, {result.removed} removed."
|
job_id, f"{result.created} created, {result.removed} removed."
|
||||||
|
)
|
||||||
|
|
||||||
job = jobs.start("fit_recompute_part", _run)
|
job = await jobs.start("fit_recompute_part", _run)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
request,
|
request,
|
||||||
"admin/_job_progress.html",
|
"admin/_job_progress.html",
|
||||||
@@ -2527,7 +2598,7 @@ async def augmented_part_fits_recompute_job_status(
|
|||||||
db: AsyncSession = Depends(get_db),
|
db: AsyncSession = Depends(get_db),
|
||||||
actor: User = Depends(web_require("fits:write")),
|
actor: User = Depends(web_require("fits:write")),
|
||||||
):
|
):
|
||||||
job = jobs.get(job_id)
|
job = await jobs.get(db, job_id)
|
||||||
if job is None:
|
if job is None:
|
||||||
ctx = await _augmented_part_fits_ctx(
|
ctx = await _augmented_part_fits_ctx(
|
||||||
db,
|
db,
|
||||||
@@ -2774,7 +2845,7 @@ async def recompute_all_fits(
|
|||||||
):
|
):
|
||||||
actor_id = actor.id
|
actor_id = actor.id
|
||||||
|
|
||||||
async def _run(job: jobs.Job) -> None:
|
async def _run(job_id: str) -> None:
|
||||||
async with AsyncSessionLocal() as db:
|
async with AsyncSessionLocal() as db:
|
||||||
result = await fit_matching_service.recompute_all(db)
|
result = await fit_matching_service.recompute_all(db)
|
||||||
actor_row = await user_crud.get(db, actor_id)
|
actor_row = await user_crud.get(db, actor_id)
|
||||||
@@ -2789,13 +2860,15 @@ async def recompute_all_fits(
|
|||||||
f"{result.removed} removed"
|
f"{result.removed} removed"
|
||||||
),
|
),
|
||||||
)
|
)
|
||||||
job.status = "done"
|
await jobs.mark_done(
|
||||||
job.message = (
|
job_id,
|
||||||
|
message=(
|
||||||
f"Recompute complete: {result.created} created, "
|
f"Recompute complete: {result.created} created, "
|
||||||
f"{result.removed} removed."
|
f"{result.removed} removed."
|
||||||
|
),
|
||||||
)
|
)
|
||||||
|
|
||||||
job = jobs.start("fit_recompute_all", _run)
|
job = await jobs.start("fit_recompute_all", _run)
|
||||||
return templates.TemplateResponse(
|
return templates.TemplateResponse(
|
||||||
request,
|
request,
|
||||||
"admin/_job_progress_inline.html",
|
"admin/_job_progress_inline.html",
|
||||||
@@ -2813,9 +2886,10 @@ async def recompute_all_fits(
|
|||||||
async def recompute_all_fits_job_status(
|
async def recompute_all_fits_job_status(
|
||||||
request: Request,
|
request: Request,
|
||||||
job_id: str,
|
job_id: str,
|
||||||
|
db: AsyncSession = Depends(get_db),
|
||||||
_: User = Depends(web_require("fits:write")),
|
_: User = Depends(web_require("fits:write")),
|
||||||
):
|
):
|
||||||
job = jobs.get(job_id)
|
job = await jobs.get(db, job_id)
|
||||||
if job is None:
|
if job is None:
|
||||||
return HTMLResponse("Recompute job not found — it may have expired.")
|
return HTMLResponse("Recompute job not found — it may have expired.")
|
||||||
if job.status == "running":
|
if job.status == "running":
|
||||||
|
|||||||
@@ -2,7 +2,11 @@ import asyncio
|
|||||||
import re
|
import re
|
||||||
|
|
||||||
import pytest
|
import pytest
|
||||||
|
import pytest_asyncio
|
||||||
|
from sqlalchemy.ext.asyncio import async_sessionmaker, create_async_engine
|
||||||
|
|
||||||
|
from app.core.database import Base
|
||||||
|
from app.crud import imported_part_config as imported_part_config_crud
|
||||||
from app.crud import role as role_crud
|
from app.crud import role as role_crud
|
||||||
from app.models.attachment_point import AttachmentPoint
|
from app.models.attachment_point import AttachmentPoint
|
||||||
from app.models.attachment_point_name import AttachmentPointName
|
from app.models.attachment_point_name import AttachmentPointName
|
||||||
@@ -13,6 +17,25 @@ from tests.test_scope_import import _scope_row, _workbook
|
|||||||
_JOB_URL_RE = re.compile(r'hx-get="([^"]+/jobs/[^"]+)"')
|
_JOB_URL_RE = re.compile(r'hx-get="([^"]+/jobs/[^"]+)"')
|
||||||
|
|
||||||
|
|
||||||
|
@pytest_asyncio.fixture
|
||||||
|
async def db_sessionmaker(tmp_path):
|
||||||
|
"""Override conftest's in-memory+StaticPool fixture for this module.
|
||||||
|
|
||||||
|
These tests run a real background asyncio task (the job) concurrently
|
||||||
|
with polling requests, both hitting the DB at once. StaticPool pins every
|
||||||
|
session to one shared raw connection, which can't safely serve two
|
||||||
|
callers at the same time; a real file-backed DB with its own connection
|
||||||
|
pool (how SQLite/Postgres actually behave outside tests) can.
|
||||||
|
"""
|
||||||
|
db_path = tmp_path / "jobs_test.db"
|
||||||
|
engine = create_async_engine(f"sqlite+aiosqlite:///{db_path}")
|
||||||
|
async with engine.begin() as conn:
|
||||||
|
await conn.run_sync(Base.metadata.create_all)
|
||||||
|
maker = async_sessionmaker(engine, expire_on_commit=False)
|
||||||
|
yield maker
|
||||||
|
await engine.dispose()
|
||||||
|
|
||||||
|
|
||||||
async def _web_login(client, email, password="secret123"):
|
async def _web_login(client, email, password="secret123"):
|
||||||
resp = await client.post("/login", data={"email": email, "password": password})
|
resp = await client.post("/login", data={"email": email, "password": password})
|
||||||
assert resp.status_code == 303
|
assert resp.status_code == 303
|
||||||
@@ -39,6 +62,7 @@ async def test_scope_import_runs_in_background(
|
|||||||
client, seeded, make_user, db_sessionmaker, monkeypatch
|
client, seeded, make_user, db_sessionmaker, monkeypatch
|
||||||
):
|
):
|
||||||
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
||||||
|
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
|
||||||
async with db_sessionmaker() as db:
|
async with db_sessionmaker() as db:
|
||||||
perms = seeded["perms"]
|
perms = seeded["perms"]
|
||||||
editor = await role_crud.create(
|
editor = await role_crud.create(
|
||||||
@@ -69,6 +93,7 @@ async def test_augmented_parts_sync_runs_in_background(
|
|||||||
client, seeded, make_user, db_sessionmaker, monkeypatch
|
client, seeded, make_user, db_sessionmaker, monkeypatch
|
||||||
):
|
):
|
||||||
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
||||||
|
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
|
||||||
async with db_sessionmaker() as db:
|
async with db_sessionmaker() as db:
|
||||||
perms = seeded["perms"]
|
perms = seeded["perms"]
|
||||||
editor = await role_crud.create(
|
editor = await role_crud.create(
|
||||||
@@ -93,6 +118,7 @@ async def test_fit_recompute_for_part_runs_in_background(
|
|||||||
client, seeded, make_user, db_sessionmaker, monkeypatch
|
client, seeded, make_user, db_sessionmaker, monkeypatch
|
||||||
):
|
):
|
||||||
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
||||||
|
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
|
||||||
async with db_sessionmaker() as db:
|
async with db_sessionmaker() as db:
|
||||||
apn = AttachmentPointName(name="Objective")
|
apn = AttachmentPointName(name="Objective")
|
||||||
part_t = PartType(name="Scope")
|
part_t = PartType(name="Scope")
|
||||||
@@ -137,6 +163,7 @@ async def test_fit_recompute_all_runs_in_background(
|
|||||||
client, seeded, make_user, db_sessionmaker, monkeypatch
|
client, seeded, make_user, db_sessionmaker, monkeypatch
|
||||||
):
|
):
|
||||||
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
||||||
|
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
|
||||||
async with db_sessionmaker() as db:
|
async with db_sessionmaker() as db:
|
||||||
perms = seeded["perms"]
|
perms = seeded["perms"]
|
||||||
editor = await role_crud.create(
|
editor = await role_crud.create(
|
||||||
@@ -154,3 +181,64 @@ async def test_fit_recompute_all_runs_in_background(
|
|||||||
|
|
||||||
final_body = await _poll_until_done(client, match.group(1))
|
final_body = await _poll_until_done(client, match.group(1))
|
||||||
assert "Recompute complete" in final_body
|
assert "Recompute complete" in final_body
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_imported_parts_refresh_runs_in_background(
|
||||||
|
client, seeded, make_user, db_sessionmaker, monkeypatch
|
||||||
|
):
|
||||||
|
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
||||||
|
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
|
||||||
|
|
||||||
|
async def _fake_fetch(url, username, password):
|
||||||
|
return [{"InventoryID": "INV-1", "Description": "Widget"}]
|
||||||
|
|
||||||
|
monkeypatch.setattr("app.web.admin.odata_service.fetch", _fake_fetch)
|
||||||
|
|
||||||
|
async with db_sessionmaker() as db:
|
||||||
|
await imported_part_config_crud.upsert(
|
||||||
|
db, odata_url="https://example.test/odata", username=None, password=None
|
||||||
|
)
|
||||||
|
perms = seeded["perms"]
|
||||||
|
editor = await role_crud.create(
|
||||||
|
db,
|
||||||
|
name="imported-parts-editor",
|
||||||
|
permission_ids=[perms["imported_parts:write"].id],
|
||||||
|
)
|
||||||
|
editor_id = editor.id
|
||||||
|
await make_user("refresher@example.com", role_ids=[editor_id])
|
||||||
|
await _web_login(client, "refresher@example.com")
|
||||||
|
|
||||||
|
resp = await client.post("/admin/imported-parts/refresh")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert 'id="imported-parts-section"' in resp.text
|
||||||
|
match = _JOB_URL_RE.search(resp.text)
|
||||||
|
assert match, resp.text
|
||||||
|
|
||||||
|
final_body = await _poll_until_done(client, match.group(1))
|
||||||
|
assert "Sync complete: 1 created, 0 updated." in final_body
|
||||||
|
assert "INV-1" in final_body
|
||||||
|
|
||||||
|
|
||||||
|
@pytest.mark.asyncio
|
||||||
|
async def test_imported_parts_refresh_without_config_skips_job(
|
||||||
|
client, seeded, make_user, db_sessionmaker, monkeypatch
|
||||||
|
):
|
||||||
|
monkeypatch.setattr("app.web.admin.AsyncSessionLocal", db_sessionmaker)
|
||||||
|
monkeypatch.setattr("app.services.jobs.AsyncSessionLocal", db_sessionmaker)
|
||||||
|
async with db_sessionmaker() as db:
|
||||||
|
perms = seeded["perms"]
|
||||||
|
editor = await role_crud.create(
|
||||||
|
db,
|
||||||
|
name="imported-parts-editor-noconfig",
|
||||||
|
permission_ids=[perms["imported_parts:write"].id],
|
||||||
|
)
|
||||||
|
editor_id = editor.id
|
||||||
|
await make_user("refresher2@example.com", role_ids=[editor_id])
|
||||||
|
await _web_login(client, "refresher2@example.com")
|
||||||
|
|
||||||
|
resp = await client.post("/admin/imported-parts/refresh")
|
||||||
|
assert resp.status_code == 200
|
||||||
|
assert "OData URL is not configured." in resp.text
|
||||||
|
# No background job started for this synchronous validation failure.
|
||||||
|
assert _JOB_URL_RE.search(resp.text) is None
|
||||||
@@ -132,20 +132,17 @@ async def test_purge_wipes_domain_data_but_keeps_users_and_odata_config(
|
|||||||
FitNote,
|
FitNote,
|
||||||
ImportedPart,
|
ImportedPart,
|
||||||
ScopeModel,
|
ScopeModel,
|
||||||
SubPartType,
|
|
||||||
PartType,
|
|
||||||
AttachmentPointName,
|
AttachmentPointName,
|
||||||
IsoThread,
|
IsoThread,
|
||||||
UncThread,
|
UncThread,
|
||||||
ClampConnection,
|
ClampConnection,
|
||||||
StaticConnection,
|
StaticConnection,
|
||||||
PartColour,
|
|
||||||
Manufacturer,
|
Manufacturer,
|
||||||
):
|
):
|
||||||
rows = (await db.execute(Model.__table__.select())).all()
|
rows = (await db.execute(Model.__table__.select())).all()
|
||||||
assert rows == [], f"{Model.__name__} should be empty after purge"
|
assert rows == [], f"{Model.__name__} should be empty after purge"
|
||||||
|
|
||||||
# Kept: users, RBAC, and the OData connection config.
|
# Kept: users, RBAC, the OData connection config, and part taxonomy.
|
||||||
from app.crud import user as user_crud
|
from app.crud import user as user_crud
|
||||||
|
|
||||||
assert await user_crud.get_by_email(db, "root@example.com") is not None
|
assert await user_crud.get_by_email(db, "root@example.com") is not None
|
||||||
@@ -153,5 +150,9 @@ async def test_purge_wipes_domain_data_but_keeps_users_and_odata_config(
|
|||||||
assert config is not None
|
assert config is not None
|
||||||
assert config.odata_url == "https://example.test/odata"
|
assert config.odata_url == "https://example.test/odata"
|
||||||
|
|
||||||
|
for Model in (PartColour, PartType, SubPartType):
|
||||||
|
rows = (await db.execute(Model.__table__.select())).all()
|
||||||
|
assert rows != [], f"{Model.__name__} should survive the purge"
|
||||||
|
|
||||||
entries = await audit_crud.list_recent(db)
|
entries = await audit_crud.list_recent(db)
|
||||||
assert any("Purged all domain data" in e.description for e in entries)
|
assert any("Purged all domain data" in e.description for e in entries)
|
||||||
@@ -275,7 +275,7 @@ async def test_scopes_page_renders_objective_attachment_point_thread(
|
|||||||
assert resp.status_code == 200
|
assert resp.status_code == 200
|
||||||
body = resp.text
|
body = resp.text
|
||||||
# The attachment point's thread spec wins over ScopeInfo.objective_thread_pitch.
|
# The attachment point's thread spec wins over ScopeInfo.objective_thread_pitch.
|
||||||
assert "M44×0.5" in body
|
assert "M44.0×0.5" in body
|
||||||
assert "M52x0.75" not in body
|
assert "M52x0.75" not in body
|
||||||
|
|
||||||
|
|
||||||
|
|||||||
Reference in new issue
Block a user