130 lines
4.5 KiB
Python
130 lines
4.5 KiB
Python
"""Feature access. Identity is Auth; this only answers 'may this profile use feature X?'."""
|
|
from __future__ import annotations
|
|
|
|
from datetime import datetime, timezone
|
|
|
|
from db import get_db, row_to_dict
|
|
|
|
|
|
class EntitlementError(Exception):
|
|
def __init__(self, code: str, message: str, status_code: int = 403):
|
|
super().__init__(message)
|
|
self.code = code
|
|
self.message = message
|
|
self.status_code = status_code
|
|
|
|
|
|
def _period_key(reset_period: str) -> str:
|
|
now = datetime.now(timezone.utc)
|
|
if reset_period == "day":
|
|
return now.strftime("%Y-%m-%d")
|
|
if reset_period == "month":
|
|
return now.strftime("%Y-%m")
|
|
return "all"
|
|
|
|
|
|
def _effective_limit(profile_id: str, feature_id: str) -> tuple[dict, int | None]:
|
|
with get_db() as conn:
|
|
feature = row_to_dict(
|
|
conn.execute("SELECT * FROM features WHERE id = ? AND active = 1", (feature_id,)).fetchone()
|
|
)
|
|
if not feature:
|
|
raise EntitlementError("unknown_feature", f"Feature ist nicht registriert: {feature_id}")
|
|
override = row_to_dict(
|
|
conn.execute(
|
|
"SELECT limit_value FROM user_feature_restrictions WHERE profile_id = ? AND feature_id = ?",
|
|
(profile_id, feature_id),
|
|
).fetchone()
|
|
)
|
|
if override is not None:
|
|
return feature, override["limit_value"]
|
|
profile = row_to_dict(
|
|
conn.execute("SELECT tier_id FROM profiles WHERE id = ?", (profile_id,)).fetchone()
|
|
)
|
|
tier_id = (profile or {}).get("tier_id") or "local"
|
|
tier_limit = row_to_dict(
|
|
conn.execute(
|
|
"SELECT limit_value FROM tier_limits WHERE tier_id = ? AND feature_id = ?",
|
|
(tier_id, feature_id),
|
|
).fetchone()
|
|
)
|
|
if tier_limit is not None:
|
|
return feature, tier_limit["limit_value"]
|
|
return feature, feature["default_limit"]
|
|
|
|
|
|
def check_feature_access(profile_id: str, feature_id: str) -> dict:
|
|
feature, limit = _effective_limit(profile_id, feature_id)
|
|
period = _period_key(feature["reset_period"])
|
|
with get_db() as conn:
|
|
usage = row_to_dict(
|
|
conn.execute(
|
|
"""
|
|
SELECT used FROM user_feature_usage
|
|
WHERE profile_id = ? AND feature_id = ? AND period_key = ?
|
|
""",
|
|
(profile_id, feature_id, period),
|
|
).fetchone()
|
|
)
|
|
used = int((usage or {}).get("used") or 0)
|
|
if feature["limit_type"] == "boolean":
|
|
allowed = limit is None or int(limit) >= 1
|
|
else:
|
|
allowed = limit is None or used < int(limit)
|
|
if not allowed:
|
|
raise EntitlementError("feature_limit", f"Kontingent erschöpft: {feature_id}")
|
|
return {
|
|
"feature_id": feature_id,
|
|
"allowed": True,
|
|
"limit": limit,
|
|
"used": used,
|
|
"period_key": period,
|
|
}
|
|
|
|
|
|
def increment_feature_usage(profile_id: str, feature_id: str) -> None:
|
|
feature, _limit = _effective_limit(profile_id, feature_id)
|
|
period = _period_key(feature["reset_period"])
|
|
with get_db() as conn:
|
|
conn.execute(
|
|
"""
|
|
INSERT INTO user_feature_usage (profile_id, feature_id, period_key, used)
|
|
VALUES (?, ?, ?, 1)
|
|
ON CONFLICT(profile_id, feature_id, period_key)
|
|
DO UPDATE SET used = used + 1
|
|
""",
|
|
(profile_id, feature_id, period),
|
|
)
|
|
|
|
|
|
def subscription_for(profile_id: str) -> dict:
|
|
with get_db() as conn:
|
|
profile = row_to_dict(
|
|
conn.execute(
|
|
"""
|
|
SELECT p.id, p.tier_id, t.name AS tier_name, t.description AS tier_description
|
|
FROM profiles p
|
|
JOIN tiers t ON t.id = p.tier_id
|
|
WHERE p.id = ?
|
|
""",
|
|
(profile_id,),
|
|
).fetchone()
|
|
)
|
|
features = [row_to_dict(row) for row in conn.execute("SELECT * FROM features WHERE active = 1 ORDER BY id")]
|
|
entitlements = []
|
|
for feature in features:
|
|
try:
|
|
entitlements.append(check_feature_access(profile_id, feature["id"]))
|
|
except EntitlementError as exc:
|
|
entitlements.append({"feature_id": feature["id"], "allowed": False, "code": exc.code})
|
|
return {
|
|
"profile_id": profile_id,
|
|
"tier": {
|
|
"id": (profile or {}).get("tier_id"),
|
|
"name": (profile or {}).get("tier_name"),
|
|
"description": (profile or {}).get("tier_description"),
|
|
},
|
|
"features": entitlements,
|
|
"billing": None,
|
|
}
|