mirror of
https://github.com/StanGirard/quivr.git
synced 2024-12-19 17:01:41 +03:00
0b091bd8c9
* ♻️ use Single Responsibility Principle on brains_subscription * ✨ new brain subscription endpoints for invited user * 📝 add documentation to endpoints * 🎨 add base_frontend_url to send custom url for brain share * ✏️ brains instead of brain in url * ✨ use origin in header for frontend url in subscription email * 🚚 move and remove unused code * ✨ new subscription API for BE endpoints in frontend * ✨ new addBrain to add a shared brain in frontend * 🥚 new hook for brain invitations * ✨ new page for brain invitation * ✨ change frontend url to copy for brain subscription * ✏️ call RBAC with wrapper function * ✏️ last typos
67 lines
1.8 KiB
Python
67 lines
1.8 KiB
Python
from typing import Optional
|
|
from uuid import UUID
|
|
|
|
import resend
|
|
from logger import get_logger
|
|
from models.settings import BrainSettings, CommonsDep, common_dependencies
|
|
from pydantic import BaseModel
|
|
|
|
logger = get_logger(__name__)
|
|
|
|
|
|
class BrainSubscription(BaseModel):
|
|
brain_id: UUID
|
|
email: str
|
|
rights: str = "Viewer"
|
|
|
|
class Config:
|
|
arbitrary_types_allowed = True
|
|
|
|
@property
|
|
def commons(self) -> CommonsDep:
|
|
return common_dependencies()
|
|
|
|
def create_subscription_invitation(self):
|
|
logger.info("Creating subscription invitation")
|
|
response = (
|
|
self.commons["supabase"]
|
|
.table("brain_subscription_invitations")
|
|
.insert(
|
|
{
|
|
"brain_id": str(self.brain_id),
|
|
"email": self.email,
|
|
"rights": self.rights,
|
|
}
|
|
)
|
|
.execute()
|
|
)
|
|
return response.data
|
|
|
|
def update_subscription_invitation(self):
|
|
logger.info("Updating subscription invitation")
|
|
response = (
|
|
self.commons["supabase"]
|
|
.table("brain_subscription_invitations")
|
|
.update({"rights": self.rights})
|
|
.eq("brain_id", str(self.brain_id))
|
|
.eq("email", self.email)
|
|
.execute()
|
|
)
|
|
return response.data
|
|
|
|
def create_or_update_subscription_invitation(self):
|
|
response = (
|
|
self.commons["supabase"]
|
|
.table("brain_subscription_invitations")
|
|
.select("*")
|
|
.eq("brain_id", str(self.brain_id))
|
|
.eq("email", self.email)
|
|
.execute()
|
|
)
|
|
|
|
if response.data:
|
|
response = self.update_subscription_invitation()
|
|
else:
|
|
response = self.create_subscription_invitation()
|
|
|
|
return response |