116 lines
4.0 KiB
Python
116 lines
4.0 KiB
Python
from datetime import datetime
|
|
|
|
from sqlalchemy import JSON as JSONType
|
|
from sqlalchemy import Boolean, Column, DateTime, ForeignKey, Integer, String
|
|
from sqlalchemy.orm import relationship
|
|
|
|
from base.orm import Base, local_session
|
|
from orm.rbac import Role
|
|
|
|
|
|
class UserNotifications(Base):
|
|
__tablename__ = "user_notifications"
|
|
# id auto
|
|
user = Column(Integer, ForeignKey("user.id"))
|
|
kind = Column(String, ForeignKey("notification.kind"))
|
|
values = Column(JSONType, nullable=True) # [ <var1>, .. ]
|
|
|
|
|
|
class UserRating(Base):
|
|
__tablename__ = "user_rating"
|
|
|
|
id = None # type: ignore
|
|
rater = Column(ForeignKey("user.id"), primary_key=True, index=True)
|
|
user = Column(ForeignKey("user.id"), primary_key=True, index=True)
|
|
value = Column(Integer)
|
|
|
|
@staticmethod
|
|
def init_table():
|
|
pass
|
|
|
|
|
|
class UserRole(Base):
|
|
__tablename__ = "user_role"
|
|
|
|
id = None # type: ignore
|
|
user = Column(ForeignKey("user.id"), primary_key=True, index=True)
|
|
role = Column(ForeignKey("role.id"), primary_key=True, index=True)
|
|
|
|
|
|
class AuthorFollower(Base):
|
|
__tablename__ = "author_follower"
|
|
|
|
id = None # type: ignore
|
|
follower = Column(ForeignKey("user.id"), primary_key=True, index=True)
|
|
author = Column(ForeignKey("user.id"), primary_key=True, index=True)
|
|
createdAt = Column(
|
|
DateTime, nullable=False, default=datetime.now, comment="Created at"
|
|
)
|
|
auto = Column(Boolean, nullable=False, default=False)
|
|
|
|
|
|
class User(Base):
|
|
__tablename__ = "user"
|
|
default_user = None
|
|
|
|
email = Column(String, unique=True, nullable=False, comment="Email")
|
|
username = Column(String, nullable=False, comment="Login")
|
|
password = Column(String, nullable=True, comment="Password")
|
|
bio = Column(String, nullable=True, comment="Bio")
|
|
userpic = Column(String, nullable=True, comment="Userpic")
|
|
name = Column(String, nullable=True, comment="Display name")
|
|
slug = Column(String, unique=True, comment="User's slug")
|
|
muted = Column(Boolean, default=False)
|
|
emailConfirmed = Column(Boolean, default=False)
|
|
createdAt = Column(
|
|
DateTime, nullable=False, default=datetime.now, comment="Created at"
|
|
)
|
|
lastSeen = Column(
|
|
DateTime, nullable=False, default=datetime.now, comment="Was online at"
|
|
)
|
|
deletedAt = Column(DateTime, nullable=True, comment="Deleted at")
|
|
links = Column(JSONType, nullable=True, comment="Links")
|
|
oauth = Column(String, nullable=True)
|
|
notifications = relationship(lambda: UserNotifications)
|
|
ratings = relationship(UserRating, foreign_keys=UserRating.user)
|
|
roles = relationship(lambda: Role, secondary=UserRole.__tablename__)
|
|
oid = Column(String, nullable=True)
|
|
|
|
@staticmethod
|
|
def init_table():
|
|
with local_session() as session:
|
|
default = session.query(User).filter(User.slug == "anonymous").first()
|
|
if not default:
|
|
default_dict = {
|
|
"email": "noreply@discours.io",
|
|
"username": "noreply@discours.io",
|
|
"name": "Аноним",
|
|
"slug": "anonymous",
|
|
}
|
|
default = User.create(**default_dict)
|
|
session.add(default)
|
|
discours_dict = {
|
|
"email": "welcome@discours.io",
|
|
"username": "welcome@discours.io",
|
|
"name": "Дискурс",
|
|
"slug": "discours",
|
|
}
|
|
discours = User.create(**discours_dict)
|
|
session.add(discours)
|
|
session.commit()
|
|
User.default_user = default
|
|
|
|
async def get_permission(self):
|
|
scope = {}
|
|
for role in self.roles:
|
|
for p in role.permissions:
|
|
if p.resource not in scope:
|
|
scope[p.resource] = set()
|
|
scope[p.resource].add(p.operation)
|
|
print(scope)
|
|
return scope
|
|
|
|
|
|
# if __name__ == "__main__":
|
|
# print(User.get_permission(user_id=1)) # type: ignore
|