Log a usage event for billing tracking. Called when a member/course is added or removed.
(
org_id: int,
feature: str,
event_type: Literal["add", "remove"],
db_session: AsyncSession,
)
| 141 | |
| 142 | |
| 143 | async def log_usage_event( |
| 144 | org_id: int, |
| 145 | feature: str, |
| 146 | event_type: Literal["add", "remove"], |
| 147 | db_session: AsyncSession, |
| 148 | ): |
| 149 | """ |
| 150 | Log a usage event for billing tracking. |
| 151 | Called when a member/course is added or removed. |
| 152 | """ |
| 153 | if feature not in PLAN_BASED_FEATURES: |
| 154 | return |
| 155 | |
| 156 | # Get current actual count |
| 157 | usage_after = await _get_actual_usage(feature, org_id, db_session) |
| 158 | |
| 159 | event = UsageEvent( |
| 160 | org_id=org_id, |
| 161 | feature=feature, |
| 162 | event_type=event_type, |
| 163 | timestamp=datetime.now(), |
| 164 | usage_after=usage_after, |
| 165 | ) |
| 166 | db_session.add(event) |
| 167 | await db_session.commit() |
| 168 | |
| 169 | # Invalidate usage cache |
| 170 | _invalidate_usage_cache(org_id) |
| 171 | |
| 172 | |
| 173 | # ============================================================================ |
no test coverage detected