|
|
from datetime import datetime, timezone |
|
|
from typing import TYPE_CHECKING |
|
|
from uuid import UUID, uuid4 |
|
|
|
|
|
from sqlmodel import Field, Relationship, SQLModel |
|
|
|
|
|
if TYPE_CHECKING: |
|
|
from langflow.services.database.models.api_key import ApiKey |
|
|
from langflow.services.database.models.flow import Flow |
|
|
from langflow.services.database.models.folder import Folder |
|
|
from langflow.services.database.models.variable import Variable |
|
|
|
|
|
|
|
|
class User(SQLModel, table=True): |
|
|
id: UUID = Field(default_factory=uuid4, primary_key=True, unique=True) |
|
|
username: str = Field(index=True, unique=True) |
|
|
password: str = Field() |
|
|
profile_image: str | None = Field(default=None, nullable=True) |
|
|
is_active: bool = Field(default=False) |
|
|
is_superuser: bool = Field(default=False) |
|
|
create_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) |
|
|
updated_at: datetime = Field(default_factory=lambda: datetime.now(timezone.utc)) |
|
|
last_login_at: datetime | None = Field(default=None, nullable=True) |
|
|
api_keys: list["ApiKey"] = Relationship( |
|
|
back_populates="user", |
|
|
sa_relationship_kwargs={"cascade": "delete"}, |
|
|
) |
|
|
store_api_key: str | None = Field(default=None, nullable=True) |
|
|
flows: list["Flow"] = Relationship(back_populates="user") |
|
|
variables: list["Variable"] = Relationship( |
|
|
back_populates="user", |
|
|
sa_relationship_kwargs={"cascade": "delete"}, |
|
|
) |
|
|
folders: list["Folder"] = Relationship( |
|
|
back_populates="user", |
|
|
sa_relationship_kwargs={"cascade": "delete"}, |
|
|
) |
|
|
|
|
|
|
|
|
class UserCreate(SQLModel): |
|
|
username: str = Field() |
|
|
password: str = Field() |
|
|
|
|
|
|
|
|
class UserRead(SQLModel): |
|
|
id: UUID = Field(default_factory=uuid4) |
|
|
username: str = Field() |
|
|
profile_image: str | None = Field() |
|
|
store_api_key: str | None = Field(nullable=True) |
|
|
is_active: bool = Field() |
|
|
is_superuser: bool = Field() |
|
|
create_at: datetime = Field() |
|
|
updated_at: datetime = Field() |
|
|
last_login_at: datetime | None = Field(nullable=True) |
|
|
|
|
|
|
|
|
class UserUpdate(SQLModel): |
|
|
username: str | None = None |
|
|
profile_image: str | None = None |
|
|
password: str | None = None |
|
|
is_active: bool | None = None |
|
|
is_superuser: bool | None = None |
|
|
last_login_at: datetime | None = None |
|
|
|