coop/shared
Archived
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Decoupling audit cleanup: fix protocol gaps, remove dead APIs

Browse Source 
- Add search_posts, entry_ids_for_content, visible_entries_for_period
  to protocols and stubs
- Delete internal_api.py and factory cleanup hook (zero callers)
- Convert utils.py to utils/ package with calendar_helpers module
- Remove deleted_at check from calendar_view template (service filters)

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
This commit is contained in:
giles
2026-02-20 10:58:07 +00:00
parent 7ee8638d6e
commit e83df2f742
8 changed files with 149 additions and 159 deletions
Download Patch File Download Diff File Expand all files Collapse all files

2
browser/templates/_types/post/admin/_calendar_view.html
View File

@@ -30,7 +30,7 @@
{# Entries for this day #}
<div class="space-y-0.5">
{% for e in month_entries %}
{% if e.start_at.date() == day.date and e.deleted_at is none %}
{% if e.start_at.date() == day.date %}
{% if e.id in associated_entry_ids %}
{# Associated entry - show with delete button #}
<div class="flex items-center gap-1 text-[10px] rounded px-1 py-0.5 bg-green-200 text-green-900">

15
contracts/protocols.py
View File

@@ -5,6 +5,7 @@ implementations (Sql*Service) and no-op stubs both satisfy them.
"""
from __future__ import annotations
from datetime import datetime
from typing import Protocol, runtime_checkable
from sqlalchemy.ext.asyncio import AsyncSession
@@ -27,6 +28,10 @@ class BlogService(Protocol):
async def get_post_by_id(self, session: AsyncSession, id: int) -> PostDTO | None: ...
async def get_posts_by_ids(self, session: AsyncSession, ids: list[int]) -> list[PostDTO]: ...
async def search_posts(
self, session: AsyncSession, query: str, page: int = 1, per_page: int = 10,
) -> tuple[list[PostDTO], int]: ...
@runtime_checkable
class CalendarService(Protocol):
@@ -107,6 +112,16 @@ class CalendarService(Protocol):
self, session: AsyncSession, user_id: int, session_id: str,
) -> None: ...
async def entry_ids_for_content(
self, session: AsyncSession, content_type: str, content_id: int,
) -> set[int]: ...
async def visible_entries_for_period(
self, session: AsyncSession, calendar_id: int,
period_start: datetime, period_end: datetime,
*, user_id: int | None, is_admin: bool, session_id: str | None,
) -> list[CalendarEntryDTO]: ...
@runtime_checkable
class MarketService(Protocol):

6
infrastructure/factory.py
View File

@@ -143,12 +143,6 @@ def create_base_app(
async def _inject_base():
return await base_context()
# --- cleanup internal API client on shutdown ---
@app.after_serving
async def _close_internal_client():
from .internal_api import close_client
await close_client()
# --- event processor ---
_event_processor = EventProcessor()

152
infrastructure/internal_api.py
View File

@@ -1,152 +0,0 @@
"""
Async HTTP client for inter-app communication.
Each app exposes internal JSON API endpoints. Other apps call them
via httpx over the Docker overlay network (or localhost in dev).
URLs resolved from env vars:
INTERNAL_URL_COOP (default http://localhost:8000)
INTERNAL_URL_MARKET (default http://localhost:8001)
INTERNAL_URL_CART (default http://localhost:8002)
Session cookie forwarding: when ``forward_session=True`` the current
request's ``coop_session`` cookie is sent along so the target app can
resolve ``g.user`` / cart identity.
"""
from __future__ import annotations
import logging
import os
from typing import Any
import httpx
from quart import request as quart_request
log = logging.getLogger("internal_api")
class DictObj:
"""Thin wrapper so ``d.key`` works on dicts returned by JSON APIs.
Jinja templates use attribute access (``item.post.slug``) which
doesn't work on plain dicts. Wrapping the API response with
``dictobj()`` makes both ``item.post.slug`` and ``item["post"]["slug"]``
work identically.
"""
__slots__ = ("_data",)
def __init__(self, data: dict):
self._data = data
def __getattr__(self, name: str):
try:
v = self._data[name]
except KeyError:
raise AttributeError(name)
if isinstance(v, dict):
return DictObj(v)
return v
def get(self, key, default=None):
v = self._data.get(key, default)
if isinstance(v, dict):
return DictObj(v)
return v
def __repr__(self):
return f"DictObj({self._data!r})"
def __bool__(self):
return bool(self._data)
def dictobj(data):
"""Recursively wrap dicts (or lists of dicts) for attribute access."""
if isinstance(data, list):
return [DictObj(d) if isinstance(d, dict) else d for d in data]
if isinstance(data, dict):
return DictObj(data)
return data
_DEFAULTS = {
"coop": "http://localhost:8000",
"market": "http://localhost:8001",
"cart": "http://localhost:8002",
"events": "http://localhost:8003",
}
_client: httpx.AsyncClient | None = None
TIMEOUT = 3.0 # seconds
def _base_url(app_name: str) -> str:
env_key = f"INTERNAL_URL_{app_name.upper()}"
return os.getenv(env_key, _DEFAULTS.get(app_name, ""))
def _get_client() -> httpx.AsyncClient:
global _client
if _client is None or _client.is_closed:
_client = httpx.AsyncClient(timeout=TIMEOUT)
return _client
async def close_client() -> None:
"""Call from ``@app.after_serving`` to cleanly close the pool."""
global _client
if _client is not None and not _client.is_closed:
await _client.aclose()
_client = None
def _session_cookies() -> dict[str, str]:
"""Extract the shared session cookie from the incoming request."""
cookie_name = "coop_session"
try:
val = quart_request.cookies.get(cookie_name)
except RuntimeError:
# No active request context
val = None
if val:
return {cookie_name: val}
return {}
async def get(
app_name: str,
path: str,
*,
forward_session: bool = False,
params: dict | None = None,
) -> dict | list | None:
"""GET ``<app_base><path>`` and return parsed JSON, or ``None`` on failure."""
url = _base_url(app_name).rstrip("/") + path
cookies = _session_cookies() if forward_session else {}
try:
resp = await _get_client().get(url, params=params, cookies=cookies)
resp.raise_for_status()
return resp.json()
except Exception as exc:
log.warning("internal_api GET %s failed: %r", url, exc)
return None
async def post(
app_name: str,
path: str,
*,
json: Any = None,
forward_session: bool = False,
) -> dict | list | None:
"""POST ``<app_base><path>`` and return parsed JSON, or ``None`` on failure."""
url = _base_url(app_name).rstrip("/") + path
cookies = _session_cookies() if forward_session else {}
try:
resp = await _get_client().post(url, json=json, cookies=cookies)
resp.raise_for_status()
return resp.json()
except Exception as exc:
log.warning("internal_api POST %s failed: %r", url, exc)
return None

70
services/calendar_impl.py
View File

@@ -5,6 +5,7 @@ calendar-domain tables on behalf of other domains.
"""
from __future__ import annotations
from datetime import datetime
from decimal import Decimal
from sqlalchemy import select, update, func
@@ -168,6 +169,75 @@ class SqlCalendarService:
)
return set(result.scalars().all())
async def visible_entries_for_period(
self, session: AsyncSession, calendar_id: int,
period_start: datetime, period_end: datetime,
*, user_id: int | None, is_admin: bool, session_id: str | None,
) -> list[CalendarEntryDTO]:
"""Return visible entries for a calendar in a date range.
Visibility rules:
- Everyone sees confirmed entries.
- Current user/session sees their own entries (any state).
- Admins also see ordered + provisional entries for all users.
"""
# User/session entries (any state)
user_entries: list[CalendarEntry] = []
if user_id or session_id:
conditions = [
CalendarEntry.calendar_id == calendar_id,
CalendarEntry.deleted_at.is_(None),
CalendarEntry.start_at >= period_start,
CalendarEntry.start_at < period_end,
]
if user_id:
conditions.append(CalendarEntry.user_id == user_id)
elif session_id:
conditions.append(CalendarEntry.session_id == session_id)
result = await session.execute(
select(CalendarEntry).where(*conditions)
.options(selectinload(CalendarEntry.calendar))
)
user_entries = list(result.scalars().all())
# Confirmed entries for everyone
result = await session.execute(
select(CalendarEntry).where(
CalendarEntry.calendar_id == calendar_id,
CalendarEntry.state == "confirmed",
CalendarEntry.deleted_at.is_(None),
CalendarEntry.start_at >= period_start,
CalendarEntry.start_at < period_end,
).options(selectinload(CalendarEntry.calendar))
)
confirmed_entries = list(result.scalars().all())
# Admin: ordered + provisional for everyone
admin_entries: list[CalendarEntry] = []
if is_admin:
result = await session.execute(
select(CalendarEntry).where(
CalendarEntry.calendar_id == calendar_id,
CalendarEntry.state.in_(("ordered", "provisional")),
CalendarEntry.deleted_at.is_(None),
CalendarEntry.start_at >= period_start,
CalendarEntry.start_at < period_end,
).options(selectinload(CalendarEntry.calendar))
)
admin_entries = list(result.scalars().all())
# Merge, deduplicate, sort
entries_by_id: dict[int, CalendarEntry] = {}
for e in confirmed_entries:
entries_by_id[e.id] = e
for e in admin_entries:
entries_by_id[e.id] = e
for e in user_entries:
entries_by_id[e.id] = e
merged = sorted(entries_by_id.values(), key=lambda e: e.start_at or period_start)
return [_entry_to_dto(e) for e in merged]
async def associated_entries(
self, session: AsyncSession, content_type: str, content_id: int, page: int,
) -> tuple[list[CalendarEntryDTO], bool]:

9
services/stubs.py
View File

@@ -31,6 +31,9 @@ class StubBlogService:
async def get_posts_by_ids(self, session: AsyncSession, ids: list[int]) -> list[PostDTO]:
return []
async def search_posts(self, session, query, page=1, per_page=10):
return [], 0
class StubCalendarService:
async def calendars_for_container(
@@ -129,6 +132,12 @@ class StubCalendarService:
) -> None:
pass
async def entry_ids_for_content(self, session, content_type, content_id):
return set()
async def visible_entries_for_period(self, session, calendar_id, period_start, period_end, *, user_id, is_admin, session_id):
return []
class StubMarketService:
async def marketplaces_for_container(

0
utils.py → utils/__init__.py
View File

54
utils/calendar_helpers.py Normal file
View File

@@ -0,0 +1,54 @@
"""Pure calendar utility functions (no ORM dependencies).
Extracted from events/bp/calendar/services/calendar_view.py so that
blog admin (and any other app) can use them without cross-app imports.
"""
from __future__ import annotations
import calendar as pycalendar
from datetime import datetime, timezone
from quart import request
def parse_int_arg(name: str, default: int | None = None) -> int | None:
"""Parse an integer query parameter from the request."""
val = request.args.get(name, "").strip()
if not val:
return default
try:
return int(val)
except ValueError:
return default
def add_months(year: int, month: int, delta: int) -> tuple[int, int]:
"""Add (or subtract) months to a given year/month, handling year overflow."""
new_month = month + delta
new_year = year + (new_month - 1) // 12
new_month = ((new_month - 1) % 12) + 1
return new_year, new_month
def build_calendar_weeks(year: int, month: int) -> list[list[dict]]:
"""Build a calendar grid for the given year and month.
Returns a list of weeks, where each week is a list of 7 day dicts.
"""
today = datetime.now(timezone.utc).date()
cal = pycalendar.Calendar(firstweekday=0) # 0 = Monday
weeks: list[list[dict]] = []
for week in cal.monthdatescalendar(year, month):
week_days = []
for d in week:
week_days.append(
{
"date": d,
"in_month": (d.month == month),
"is_today": (d == today),
}
)
weeks.append(week_days)
return weeks