MediaRouter / app /api /social.py
basyx's picture
Upload 340 files
3493993 verified
Raw
History Blame Contribute Delete
18.6 kB
from __future__ import annotations
from datetime import datetime
from typing import Annotated
from fastapi import APIRouter, Header, HTTPException, Query, Request, Response, status
from app.social.schemas.accounts import (
SocialAccountConnectRequest,
SocialAccountSelectionRequest,
SocialAccountView,
SocialConnectResponse,
SocialProviderView,
SocialPublishOptionsView,
)
from app.social.schemas.assets import (
SocialMediaAssetRegister,
SocialMediaAssetView,
)
from app.social.schemas.jobs import SocialJobView
from app.social.schemas.operations import (
PublishingBatchView,
PublishingBulkRequest,
PublishingCalendarView,
PublishingContextView,
PublishingQueueView,
)
from app.social.schemas.posts import (
SocialPostCreate,
SocialPostDuplicateRequest,
SocialPostPatch,
SocialPostValidation,
SocialPostView,
)
from app.social.schemas.scheduling import (
SocialRescheduleRequest,
SocialScheduleCreate,
SocialScheduleView,
)
router = APIRouter(prefix="/v1/social", tags=["social automation"])
def _social(request: Request):
service = request.app.state.container.social
service.ensure_ready()
return service
def _identity(request: Request) -> tuple[str, str]:
context = request.state.auth
if not context.workspace_id or not context.user_id:
# Tenant resolution is performed by APIKeyService after credential
# verification. Never fall back to treating an API-key ID as a tenant.
raise HTTPException(status_code=403, detail="No active workspace membership.")
return context.workspace_id, context.user_id
async def _audit(
request: Request,
event_type: str,
*,
provider: str | None = None,
account_id: str | None = None,
post_id: str | None = None,
job_id: str | None = None,
) -> None:
workspace_id, _ = _identity(request)
await request.app.state.container.social.audit.record(
workspace_id=workspace_id,
event_type=event_type,
api_key_id=request.state.auth.api_key_id,
request_id=request.state.request_id,
provider=provider,
social_account_id=account_id,
social_post_id=post_id,
social_job_id=job_id,
)
@router.get("/providers", response_model=list[SocialProviderView])
async def list_providers(request: Request) -> list[SocialProviderView]:
return request.app.state.container.social.accounts.list_providers()
@router.get("/providers/{provider}/capabilities", response_model=SocialProviderView)
async def provider_capabilities(request: Request, provider: str) -> SocialProviderView:
return request.app.state.container.social.accounts.get_provider(provider)
@router.get("/assets", response_model=list[SocialMediaAssetView])
async def list_social_media_assets(
request: Request,
offset: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
) -> list[SocialMediaAssetView]:
workspace_id, _ = _identity(request)
return await _social(request).media_assets.list(workspace_id, offset=offset, limit=limit)
@router.post("/assets", response_model=SocialMediaAssetView, status_code=status.HTTP_201_CREATED)
async def register_social_media_asset(
request: Request, payload: SocialMediaAssetRegister
) -> SocialMediaAssetView:
workspace_id, _ = _identity(request)
return await _social(request).media_assets.register(workspace_id, payload)
@router.get("/accounts", response_model=list[SocialAccountView])
async def list_accounts(
request: Request,
offset: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
) -> list[SocialAccountView]:
workspace_id, _ = _identity(request)
return await _social(request).accounts.list(workspace_id, offset=offset, limit=limit)
@router.post("/accounts/select", response_model=list[SocialAccountView])
async def select_discovered_accounts(
request: Request, payload: SocialAccountSelectionRequest
) -> list[SocialAccountView]:
workspace_id, _ = _identity(request)
selected = await _social(request).accounts.select_discovered(workspace_id, payload.account_ids)
for account in selected:
await _audit(
request,
"SOCIAL_ACCOUNT_CONNECTED",
provider=account.provider.value,
account_id=account.id,
)
return selected
@router.get("/accounts/{account_id}", response_model=SocialAccountView)
async def get_account(request: Request, account_id: str) -> SocialAccountView:
workspace_id, _ = _identity(request)
return await _social(request).accounts.get(workspace_id, account_id)
@router.get(
"/accounts/{account_id}/publish-options",
response_model=SocialPublishOptionsView,
)
async def get_account_publish_options(
request: Request, account_id: str
) -> SocialPublishOptionsView:
workspace_id, _ = _identity(request)
return await _social(request).publishing.publish_options(workspace_id, account_id)
@router.post("/accounts/{provider}/connect", response_model=SocialConnectResponse)
async def connect_account(
request: Request, provider: str, payload: SocialAccountConnectRequest
) -> SocialConnectResponse:
workspace_id, user_id = _identity(request)
result = await _social(request).oauth.connect(
provider=provider,
workspace_id=workspace_id,
user_id=user_id,
payload=payload,
)
await _audit(request, "SOCIAL_ACCOUNT_CONNECTION_STARTED", provider=provider)
return result
@router.get(
"/accounts/{provider}/callback",
response_model=SocialAccountView,
include_in_schema=True,
)
async def oauth_callback(
request: Request,
provider: str,
state: Annotated[str, Query(min_length=32, max_length=255, pattern=r"^[A-Za-z0-9_-]+$")],
code: Annotated[str | None, Query(min_length=1, max_length=4096)] = None,
error: Annotated[str | None, Query(max_length=128)] = None,
) -> SocialAccountView:
# This provider-facing route is authenticated by a short-lived, single-use
# state record. Tenant/user identifiers are never accepted from the query.
social = request.app.state.container.social
social.ensure_ready()
if error or not code:
await social.oauth.callback_denied(provider=provider, state=state)
raise AssertionError("OAuth callback denial should raise a social error")
return await social.oauth.callback(provider=provider, state=state, code=code)
@router.post("/accounts/{account_id}/refresh", response_model=SocialAccountView)
async def refresh_account(request: Request, account_id: str) -> SocialAccountView:
workspace_id, _ = _identity(request)
result = await _social(request).oauth.refresh(workspace_id=workspace_id, account_id=account_id)
await _audit(request, "SOCIAL_ACCOUNT_REAUTHORIZED", account_id=account_id)
return result
@router.delete("/accounts/{account_id}", status_code=status.HTTP_204_NO_CONTENT)
async def disconnect_account(request: Request, account_id: str) -> Response:
workspace_id, _ = _identity(request)
account = await _social(request).accounts.get(workspace_id, account_id)
await request.app.state.container.social.accounts.disconnect(workspace_id, account_id)
await _audit(
request,
"SOCIAL_ACCOUNT_DISCONNECTED",
provider=account.provider.value,
account_id=account_id,
)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.post("/posts", response_model=SocialPostView, status_code=status.HTTP_201_CREATED)
async def create_post(
request: Request,
payload: SocialPostCreate,
idempotency_key: str | None = Header(
default=None, alias="Idempotency-Key", min_length=8, max_length=255
),
) -> SocialPostView:
workspace_id, user_id = _identity(request)
result = await _social(request).publishing.create(
workspace_id=workspace_id,
user_id=user_id,
payload=payload,
idempotency_key=idempotency_key,
)
await _audit(request, "SOCIAL_POST_CREATED", post_id=result.id)
if result.publish_mode.value == "draft":
await _audit(request, "publishing.draft_created", post_id=result.id)
return result
@router.get("/posts", response_model=list[SocialPostView])
async def list_posts(
request: Request,
offset: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
post_status: str | None = Query(default=None, alias="status", max_length=32),
project_id: str | None = Query(default=None, max_length=36),
search: str | None = Query(default=None, max_length=200),
) -> list[SocialPostView]:
workspace_id, _ = _identity(request)
return await _social(request).publishing.list(
workspace_id,
offset=offset,
limit=limit,
status=post_status,
project_id=project_id,
search=search,
)
@router.get("/posts/{post_id}", response_model=SocialPostView)
async def get_post(request: Request, post_id: str) -> SocialPostView:
workspace_id, _ = _identity(request)
return await _social(request).publishing.get(workspace_id, post_id)
@router.patch("/posts/{post_id}", response_model=SocialPostView)
async def update_post(request: Request, post_id: str, payload: SocialPostPatch) -> SocialPostView:
workspace_id, _ = _identity(request)
result = await _social(request).operations.update_draft(workspace_id, post_id, payload)
await _audit(request, "publishing.draft_updated", post_id=post_id)
return result
@router.get("/drafts", response_model=list[SocialPostView])
async def list_drafts(
request: Request,
offset: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
search: str | None = Query(default=None, max_length=200),
) -> list[SocialPostView]:
workspace_id, _ = _identity(request)
posts = await _social(request).publishing.list(
workspace_id,
offset=offset,
limit=limit,
search=search,
)
return [
post
for post in posts
if post.status.value in {"draft", "ready", "failed"} and post.publish_mode.value == "draft"
]
@router.patch("/drafts/{post_id}", response_model=SocialPostView)
async def update_draft(request: Request, post_id: str, payload: SocialPostPatch) -> SocialPostView:
return await update_post(request, post_id, payload)
@router.delete("/drafts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_draft(request: Request, post_id: str) -> Response:
workspace_id, _ = _identity(request)
await _social(request).operations.delete_draft(workspace_id, post_id)
await _audit(request, "publishing.draft_deleted", post_id=post_id)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.post("/posts/{post_id}/duplicate", response_model=SocialPostView)
async def duplicate_post(
request: Request,
post_id: str,
payload: SocialPostDuplicateRequest,
idempotency_key: str = Header(alias="Idempotency-Key", min_length=8, max_length=255),
) -> SocialPostView:
workspace_id, user_id = _identity(request)
result = await _social(request).operations.duplicate(
workspace_id,
user_id,
post_id,
payload,
idempotency_key=idempotency_key,
)
await _audit(request, "publishing.duplicated", post_id=result.id)
return result
@router.post("/posts/{post_id}/validate", response_model=SocialPostValidation)
async def validate_post(request: Request, post_id: str) -> SocialPostValidation:
workspace_id, _ = _identity(request)
result = await _social(request).publishing.validate_post_targets(workspace_id, post_id)
await _audit(request, "SOCIAL_POST_VALIDATED", post_id=post_id)
await _audit(request, "publishing.validated", post_id=post_id)
return result
@router.delete("/posts/{post_id}", status_code=status.HTTP_204_NO_CONTENT)
async def delete_post(request: Request, post_id: str) -> Response:
workspace_id, _ = _identity(request)
await _social(request).publishing.delete(workspace_id, post_id)
await _audit(request, "SOCIAL_POST_DELETED", post_id=post_id)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.post(
"/posts/{post_id}/publish",
response_model=list[SocialJobView],
status_code=status.HTTP_202_ACCEPTED,
)
async def publish_post(
request: Request,
post_id: str,
idempotency_key: str = Header(alias="Idempotency-Key", min_length=8, max_length=255),
) -> list[SocialJobView]:
workspace_id, _ = _identity(request)
jobs = await _social(request).publishing.queue(
workspace_id, post_id, idempotency_key=idempotency_key
)
await _audit(request, "SOCIAL_POST_PUBLISH_STARTED", post_id=post_id)
return jobs
@router.post("/posts/{post_id}/schedule", response_model=SocialScheduleView)
async def schedule_post(
request: Request, post_id: str, payload: SocialScheduleCreate
) -> SocialScheduleView:
workspace_id, _ = _identity(request)
validation = await _social(request).publishing.validate_post_targets(workspace_id, post_id)
if not validation.valid:
raise HTTPException(
status_code=422,
detail=validation.model_dump(mode="json"),
)
schedule = await _social(request).scheduling.schedule(workspace_id, post_id, payload)
await _audit(request, "SOCIAL_SCHEDULE_CREATED", post_id=post_id)
await _audit(request, "SOCIAL_POST_SCHEDULED", post_id=post_id)
await _audit(request, "publishing.scheduled", post_id=post_id)
return schedule
@router.post("/posts/{post_id}/reschedule", response_model=SocialScheduleView)
async def reschedule_post(
request: Request, post_id: str, payload: SocialRescheduleRequest
) -> SocialScheduleView:
workspace_id, _ = _identity(request)
result = await _social(request).operations.reschedule(workspace_id, post_id, payload)
await _audit(request, "publishing.rescheduled", post_id=post_id)
return result
@router.post("/posts/{post_id}/cancel", response_model=SocialPostView)
async def cancel_post(request: Request, post_id: str) -> SocialPostView:
workspace_id, _ = _identity(request)
result = await _social(request).publishing.cancel(workspace_id, post_id)
await _audit(request, "SOCIAL_POST_CANCELLED", post_id=post_id)
await _audit(request, "publishing.cancelled", post_id=post_id)
return result
@router.get("/calendar", response_model=PublishingCalendarView)
async def publishing_calendar(
request: Request,
starts_at: datetime = Query(),
ends_at: datetime = Query(),
offset: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
) -> PublishingCalendarView:
workspace_id, _ = _identity(request)
return await _social(request).operations.calendar(
workspace_id,
starts_at=starts_at,
ends_at=ends_at,
offset=offset,
limit=limit,
)
@router.get("/publishing-context", response_model=PublishingContextView)
async def publishing_context(request: Request) -> PublishingContextView:
workspace_id, _ = _identity(request)
return PublishingContextView(
timezone=await _social(request).operations.workspace_timezone(workspace_id)
)
@router.get("/queue", response_model=PublishingQueueView)
async def publishing_queue(
request: Request,
offset: int = Query(default=0, ge=0),
limit: int = Query(default=50, ge=1, le=200),
queue_status: str | None = Query(default=None, alias="status", max_length=32),
provider: str | None = Query(default=None, max_length=32),
account_id: str | None = Query(default=None, max_length=120),
project_id: str | None = Query(default=None, max_length=36),
search: str | None = Query(default=None, max_length=200),
) -> PublishingQueueView:
workspace_id, _ = _identity(request)
return await _social(request).operations.queue(
workspace_id,
offset=offset,
limit=limit,
status=queue_status,
provider=provider,
account_id=account_id,
project_id=project_id,
search=search,
)
@router.post(
"/bulk",
response_model=PublishingBatchView,
status_code=status.HTTP_202_ACCEPTED,
)
async def create_publishing_batch(
request: Request,
payload: PublishingBulkRequest,
idempotency_key: str = Header(alias="Idempotency-Key", min_length=8, max_length=255),
) -> PublishingBatchView:
workspace_id, user_id = _identity(request)
result = await _social(request).operations.create_batch(
workspace_id,
user_id,
payload,
idempotency_key=idempotency_key,
)
await _audit(request, "publishing.bulk_started")
return result
@router.post(
"/posts/{post_id}/targets/{target_id}/retry",
response_model=SocialJobView,
status_code=status.HTTP_202_ACCEPTED,
)
async def retry_post_target(
request: Request,
post_id: str,
target_id: str,
idempotency_key: str = Header(alias="Idempotency-Key", min_length=8, max_length=255),
) -> SocialJobView:
workspace_id, _ = _identity(request)
result = await _social(request).publishing.retry_target(
workspace_id,
post_id,
target_id,
idempotency_key=idempotency_key,
)
await _audit(
request,
"SOCIAL_TARGET_RETRY_QUEUED",
provider=result.provider.value if result.provider else None,
post_id=post_id,
job_id=result.id,
)
return result
@router.get("/jobs", response_model=list[SocialJobView])
async def list_jobs(
request: Request,
offset: int = Query(default=0, ge=0),
limit: int = Query(default=100, ge=1, le=500),
) -> list[SocialJobView]:
workspace_id, _ = _identity(request)
return await _social(request).jobs.list(workspace_id, offset=offset, limit=limit)
@router.get("/jobs/{job_id}", response_model=SocialJobView)
async def get_job(request: Request, job_id: str) -> SocialJobView:
workspace_id, _ = _identity(request)
return await _social(request).jobs.get(workspace_id, job_id)
@router.get("/accounts/{account_id}/analytics")
async def account_analytics(request: Request, account_id: str) -> dict[str, object]:
workspace_id, _ = _identity(request)
return await _social(request).analytics.account(workspace_id, account_id)
@router.get("/posts/{post_id}/analytics")
async def post_analytics(request: Request, post_id: str) -> dict[str, object]:
workspace_id, _ = _identity(request)
return await _social(request).analytics.post(workspace_id, post_id)