wip: redis, sqlalchemy, structured, etc

orm/__init__.py

@@ -0,0 +1,4 @@
+from orm.rbac import Operation, Permission, Role
+from orm.user import User
+
+__all__ = ["User", "Role", "Operation", "Permission"]

orm/base.py

@@ -0,0 +1,45 @@
+from typing import TypeVar, Any, Dict, Generic, Callable
+
+from sqlalchemy import create_engine, Column, Integer
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+from sqlalchemy.sql.schema import Table
+
+from settings import SQLITE_URI
+
+engine = create_engine(f'sqlite:///{SQLITE_URI}', convert_unicode=True, echo=False)
+Session = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+global_session = Session()
+
+T = TypeVar("T")
+
+REGISTRY: Dict[str, type] = {}
+
+
+class Base(declarative_base()):
+    __table__: Table
+    __tablename__: str
+    __new__: Callable
+    __init__: Callable
+
+    __abstract__: bool = True
+    __table_args__ = {"extend_existing": True}
+    id: int = Column(Integer, primary_key=True)
+    session = global_session
+
+    def __init_subclass__(cls, **kwargs):
+        REGISTRY[cls.__name__] = cls
+
+    @classmethod
+    def create(cls: Generic[T], **kwargs) -> Generic[T]:
+        instance = cls(**kwargs)
+        return instance.save()
+
+    def save(self) -> Generic[T]:
+        self.session.add(self)
+        self.session.commit()
+        return self
+
+    def dict(self) -> Dict[str, Any]:
+        column_names = self.__table__.columns.keys()
+        return {c: getattr(self, c) for c in column_names}

orm/like.py

@@ -0,0 +1,17 @@
+from typing import List
+
+from sqlalchemy import Column, Integer, String, ForeignKey, Datetime
+
+from orm import Permission
+from orm.base import Base
+
+
+class Like(Base):
+    __tablename__ = 'like'
+
+    author_id: str = Column(ForeignKey("user.id"), nullable=False, comment="Author")
+    value: str = Column(String, nullable=False, comment="Value")
+    shout: str = Column(ForeignKey("shout.id"), nullable=True, comment="Liked shout")
+    user: str = Column(ForeignKey("user.id"), nullable=True, comment="Liked user")
+
+    # TODO: add resolvers, debug, etc.

orm/message.py

@@ -0,0 +1,18 @@
+from typing import List
+
+from sqlalchemy import Column, Integer, String, ForeignKey, Datetime
+
+from orm import Permission
+from orm.base import Base
+
+
+class Message(Base):
+    __tablename__ = 'message'
+
+    sender: str = Column(ForeignKey("user.id"), nullable=False, comment="Sender")
+    body: str = Column(String, nullable=False, comment="Body")
+    createdAt: str = Column(Datetime, nullable=False, comment="Created at")
+    updatedAt: str = Column(Datetime, nullable=True, comment="Updated at")
+    replyTo: str = Column(ForeignKey("message.id", nullable=True, comment="Reply to"))
+
+    # TODO: work in progress, udpate this code

orm/proposal.py

@@ -0,0 +1,18 @@
+from typing import List
+from datetime import datetime
+from sqlalchemy import Column, Integer, String, ForeignKey, Datetime
+
+from orm import Permission
+from orm.base import Base
+
+
+class Proposal(Base):
+    __tablename__ = 'proposal'
+
+    author_id: str = Column(ForeignKey("user.id"), nullable=False, comment="Author")
+    body: str = Column(String, nullable=False, comment="Body")
+    createdAt: str = Column(datetime, nullable=False, comment="Created at")
+    shout: str = Column(ForeignKey("shout.id"), nullable=False, comment="Updated at")
+    range: str = Column(String, nullable=True, comment="Range in format <start index>:<end>")
+
+    # TODO: debug, logix

orm/rbac.py

@@ -0,0 +1,65 @@
+import warnings
+
+from typing import Type
+
+from sqlalchemy import String, Column, ForeignKey, types, UniqueConstraint
+
+from orm.base import Base, REGISTRY, engine, global_session
+
+
+class ClassType(types.TypeDecorator):
+    impl = types.String
+
+    @property
+    def python_type(self):
+        return NotImplemented
+
+    def process_literal_param(self, value, dialect):
+        return NotImplemented
+
+    def process_bind_param(self, value, dialect):
+        return value.__name__ if isinstance(value, type) else str(value)
+
+    def process_result_value(self, value, dialect):
+        class_ = REGISTRY.get(value)
+        if class_ is None:
+            warnings.warn(f"Can't find class <{value}>,find it yourself 😊", stacklevel=2)
+        return class_
+
+
+class Role(Base):
+    __tablename__ = 'role'
+    name: str = Column(String, nullable=False, unique=True, comment="Role Name")
+
+
+class Operation(Base):
+    __tablename__ = 'operation'
+    name: str = Column(String, nullable=False, unique=True, comment="Operation Name")
+
+
+class Resource(Base):
+    __tablename__ = "resource"
+    resource_class: Type[Base] = Column(ClassType, nullable=False, unique=True, comment="Resource class")
+    name: str = Column(String, nullable=False, unique=True, comment="Resource name")
+
+
+class Permission(Base):
+    __tablename__ = "permission"
+    __table_args__ = (UniqueConstraint("role_id", "operation_id", "resource_id"), {"extend_existing": True})
+
+    role_id: int = Column(ForeignKey("role.id", ondelete="CASCADE"), nullable=False, comment="Role")
+    operation_id: int = Column(ForeignKey("operation.id", ondelete="CASCADE"), nullable=False, comment="Operation")
+    resource_id: int = Column(ForeignKey("operation.id", ondelete="CASCADE"), nullable=False, comment="Resource")
+
+
+if __name__ == '__main__':
+    Base.metadata.create_all(engine)
+    ops = [
+        Permission(role_id=1, operation_id=1, resource_id=1),
+        Permission(role_id=1, operation_id=2, resource_id=1),
+        Permission(role_id=1, operation_id=3, resource_id=1),
+        Permission(role_id=1, operation_id=4, resource_id=1),
+        Permission(role_id=2, operation_id=4, resource_id=1)
+    ]
+    global_session.add_all(ops)
+    global_session.commit()

orm/shout.py

@@ -0,0 +1,17 @@
+from typing import List
+from datetime import datetime
+from sqlalchemy import Column, Integer, String, ForeignKey, Datetime
+
+from orm import Permission
+from orm.base import Base
+
+
+class Shout(Base):
+    __tablename__ = 'shout'
+
+    author_id: str = Column(ForeignKey("user.id"), nullable=False, comment="Author")
+    body: str = Column(String, nullable=False, comment="Body")
+    createdAt: str = Column(datetime, nullable=False, comment="Created at")
+    updatedAt: str = Column(datetime, nullable=False, comment="Updated at")
+
+    # TODO: add all the fields

orm/user.py

@@ -0,0 +1,26 @@
+from typing import List
+
+from sqlalchemy import Column, Integer, String, ForeignKey
+
+from orm import Permission
+from orm.base import Base
+
+
+class User(Base):
+    __tablename__ = 'user'
+
+    name: str = Column(String, nullable=False, comment="Name")
+    password: str = Column(String, nullable=False, comment="Password")
+    # phone: str = Column(String, comment="Phone")
+    # age: int = Column(Integer, comment="Age")
+    role_id: int = Column(ForeignKey("role.id"), nullable=False, comment="Role")
+
+    @classmethod
+    def get_permission(cls, user_id):
+        perms: List[Permission] = cls.session.query(Permission).join(User, User.role_id == Permission.role_id).filter(
+            User.id == user_id).all()
+        return {f"{p.operation_id}-{p.resource_id}" for p in perms}
+
+
+if __name__ == '__main__':
+    print(User.get_permission(user_id=1))