stage

bot/db/engine.py

@@ -0,0 +1,19 @@
+from sqlalchemy import create_engine
+from sqlalchemy.ext.declarative import declarative_base
+from sqlalchemy.orm import sessionmaker
+import os
+
+POSTGRES_USER = os.environ.get("POSTGRES_USER", "user")
+POSTGRES_PASSWORD = os.environ.get("POSTGRES_PASSWORD", "password")
+POSTGRES_DB = os.environ.get("POSTGRES_DB", "db")
+POSTGRES_HOST = os.environ.get("POSTGRES_HOST", "postgres")
+
+
+SQLALCHEMY_DATABASE_URL = (
+    f"postgresql://{POSTGRES_USER}:{POSTGRES_PASSWORD}@{POSTGRES_HOST}/{POSTGRES_DB}"
+)
+
+engine = create_engine(SQLALCHEMY_DATABASE_URL)
+SessionLocal = sessionmaker(autocommit=False, autoflush=False, bind=engine)
+
+Base = declarative_base()

bot/db/models.py

@@ -0,0 +1,46 @@
+from sqlalchemy import (
+    Column,
+    String,
+    types,
+)
+from sqlalchemy.dialects.postgresql import UUID
+import uuid
+from typing import Literal
+
+
+from .database import Base
+
+# DOMAIN_URL = os.environ.get("DOMAIN_URL", "localhost")
+
+StatusType = Literal["offline", "online", "webhook_error", "token_error"]
+
+
+class ChoiceType(types.TypeDecorator):
+    impl = types.String
+
+    def __init__(self, choices, **kw):
+        self.choices = dict(choices)
+        super(ChoiceType, self).__init__(**kw)
+
+    def process_bind_param(self, value, dialect):
+        return [k for k, v in self.choices.items() if v == value][0]
+
+    def process_result_value(self, value, dialect):
+        return self.choices[value]
+
+
+class TelegramUpload(Base):
+    __tablename__ = "telegramuploads"
+
+    id = Column(UUID(as_uuid=True), primary_key=True, default=uuid.uuid4)
+    path = Column(String, index=True)
+    file_id = Column(String, index=True, default=None)
+    type = Column(
+        ChoiceType(
+            {
+                "pic": "pic",
+                "gif": "gif",
+            }
+        ),
+        default=None,
+    )