feat: add multi account (#5)

.env.example

@@ -3,4 +3,5 @@ TELEGRAM_BOT_TOKEN="123456789:XXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
 TELEGRAM_ADMINS=[123456789,987654321]
 
 ### HEYZNER API KEY SETTINGS
-HETZNER_API_KEY="XXXXXXXXXXXXXXXXXXXXXXXXXXXXX"
+## if you one more api key, you can add like this [ ["name1","key1"] , ["name2","key2"] ]
+HETZNER_API_KEYS=[ ["name","api key"] ]

api/__init__.py

@@ -1,6 +1,3 @@
 from .hetzner import HetznerManager
-from config import EnvFile
 
-HetznerAPI = HetznerManager(key=EnvFile.HETZNER_API_KEY)
-
-__all__ = ["HetznerAPI"]
+__all__ = ["HetznerManager"]

api/hetzner.py

@@ -33,74 +33,102 @@ async def wrapper(*args, **kwargs):
 
 
 class HetznerManager:
-    def __init__(self, key: str) -> None:
-        self.client = Client(token=key)
-
+    @staticmethod
     @handle_hetzner_errors
-    async def get_servers(self) -> List[Server]:
-        return self.client.servers.get_all()
+    async def get_servers(key: str) -> List[Server]:
+        client = Client(token=key)
+        return client.servers.get_all()
 
+    @staticmethod
     @handle_hetzner_errors
-    async def get_server(self, server_id: int) -> Optional[Server]:
-        return self.client.servers.get_by_id(server_id)
+    async def get_server(key: str, server_id: int) -> Optional[Server]:
+        client = Client(token=key)
+        return client.servers.get_by_id(server_id)
 
+    @staticmethod
     @handle_hetzner_errors
-    async def get_server_type(self, server_id: int) -> Optional[ServerType]:
-        return self.client.server_types.get_by_id(server_id)
+    async def get_server_type(key: str, server_id: int) -> Optional[ServerType]:
+        client = Client(token=key)
+        return client.server_types.get_by_id(server_id)
 
+    @staticmethod
     @handle_hetzner_errors
-    async def power_on(self, server: Server) -> Optional[Action]:
-        return self.client.servers.power_on(server)
+    async def power_on(key: str, server: Server) -> Optional[Action]:
+        client = Client(token=key)
+        return client.servers.power_on(server)
 
+    @staticmethod
     @handle_hetzner_errors
-    async def power_off(self, server: Server) -> Optional[Action]:
-        return self.client.servers.power_off(server)
+    async def power_off(key: str, server: Server) -> Optional[Action]:
+        client = Client(token=key)
+        return client.servers.power_off(server)
 
+    @staticmethod
     @handle_hetzner_errors
-    async def reboot(self, server: Server) -> Optional[Action]:
-        return self.client.servers.reboot(server)
+    async def reboot(key: str, server: Server) -> Optional[Action]:
+        client = Client(token=key)
+        return client.servers.reboot(server)
 
+    @staticmethod
     @handle_hetzner_errors
-    async def reset_password(self, server: Server) -> Optional[str]:
-        return self.client.servers.reset_password(server).root_password
+    async def reset_password(key: str, server: Server) -> Optional[str]:
+        client = Client(token=key)
+        return client.servers.reset_password(server).root_password
 
+    @staticmethod
     @handle_hetzner_errors
-    async def delete(self, server: Server) -> Optional[Action]:
-        return self.client.servers.delete(server)
+    async def delete(key: str, server: Server) -> Optional[Action]:
+        client = Client(token=key)
+        return client.servers.delete(server)
 
+    @staticmethod
     @handle_hetzner_errors
-    async def reset(self, server: Server) -> Optional[Action]:
-        return self.client.servers.reset(server)
+    async def reset(key: str, server: Server) -> Optional[Action]:
+        client = Client(token=key)
+        return client.servers.reset(server)
 
+    @staticmethod
     @handle_hetzner_errors
-    async def get_images(self, arch: str = None) -> List[Image]:
-        return self.client.images.get_all(architecture=arch)
+    async def get_images(key: str, arch: str = None) -> List[Image]:
+        client = Client(token=key)
+        return client.images.get_all(architecture=arch)
 
+    @staticmethod
     @handle_hetzner_errors
-    async def rebuild_server(self, server: Server, image_id: int) -> Optional[Action]:
-        image = self.client.images.get_by_id(image_id)
-        return self.client.servers.rebuild(server, image)
-
+    async def rebuild_server(
+        key: str, server: Server, image_id: int
+    ) -> Optional[Action]:
+        client = Client(token=key)
+        image = client.images.get_by_id(image_id)
+        return client.servers.rebuild(server, image)
+
+    @staticmethod
     @handle_hetzner_errors
     async def create_server(
-        self, name: str, server_type: ServerType, image: Image
+        key: str, name: str, server_type: ServerType, image: Image
     ) -> Optional[Server]:
-        server = self.client.servers.create(
-            name=name, server_type=server_type, image=image
-        )
+        client = Client(token=key)
+        server = client.servers.create(name=name, server_type=server_type, image=image)
         return server.server
 
-    async def get_image(self, id: int) -> Optional[Image]:
-        return self.client.images.get_by_id(id)
+    async def get_image(key: str, id: int) -> Optional[Image]:
+        client = Client(token=key)
+        return client.images.get_by_id(id)
 
+    @staticmethod
     @handle_hetzner_errors
-    async def get_server_types(self) -> List[ServerType]:
-        return self.client.server_types.get_all()
+    async def get_server_types(key: str) -> List[ServerType]:
+        client = Client(token=key)
+        return client.server_types.get_all()
 
+    @staticmethod
     @handle_hetzner_errors
-    async def get_datacenters(self) -> List[Datacenter]:
-        return self.client.datacenters.get_all()
+    async def get_datacenters(key: str) -> List[Datacenter]:
+        client = Client(token=key)
+        return client.datacenters.get_all()
 
+    @staticmethod
     @handle_hetzner_errors
-    async def get_datacenter(self, id: int) -> Optional[Datacenter]:
-        return self.client.datacenters.get_by_id(id)
+    async def get_datacenter(key: str, id: int) -> Optional[Datacenter]:
+        client = Client(token=key)
+        return client.datacenters.get_by_id(id)

config/env.py

@@ -1,3 +1,4 @@
+import hashlib
 from pydantic_settings import BaseSettings, SettingsConfigDict
 
 
@@ -10,8 +11,22 @@ class EnvFileReader(BaseSettings):
 
     TELEGRAM_BOT_TOKEN: str = ""
     TELEGRAM_ADMINS: list[int] = []
-    HETZNER_API_KEY: str = ""
+    HETZNER_API_KEYS: list[tuple[str, str]] = []
 
     def is_admin(self, userid: int) -> bool:
         """check userid is admin or not"""
         return userid in self.TELEGRAM_ADMINS
+
+    def to_hash(self, key: str, length: int = 4) -> str:
+        """create a hash from key"""
+        hash_object = hashlib.md5(key.encode())
+        full_hash = hash_object.hexdigest()
+        return full_hash[:length]
+
+    def from_hash(self, hashkey: str, length: int = 4) -> str:
+        """create a key from hash"""
+        for key_pair in self.HETZNER_API_KEYS:
+            current_hash = self.to_hash(key_pair[1])
+            if current_hash == hashkey:
+                return key_pair[1]
+        return None

keys/__init__.py

@@ -1,21 +1,15 @@
-from .action import Actions
-from .callback import (
-    ServerAction,
-    ServerList,
-    ServerTypeSelect,
-    LocationTypeSelect,
-    ImageTypeSelect,
-)
+from .enums import Actions, Pages, ServerCreate, ServerUpdate
+from .callback import PageCB, SelectCB
 from .keyboard import KeyboardsCreater
 
 Keyboards = KeyboardsCreater()
 
 __all__ = [
     "Actions",
-    "ServerAction",
-    "ServerList",
+    "PageCB",
+    "SelectCB",
+    "Pages",
     "Keyboards",
-    "ServerTypeSelect",
-    "LocationTypeSelect",
-    "ImageTypeSelect",
+    "ServerCreate",
+    "ServerUpdate",
 ]

keys/callback.py

@@ -1,27 +1,19 @@
 from aiogram.filters.callback_data import CallbackData
+from .enums import Pages, Actions, ServerCreate, ServerUpdate
 
 
-class ServerAction(CallbackData, prefix="server_action"):
-    action: str
-    server_id: int
+class PageCB(CallbackData, prefix="pages"):
+    key: str = "KEY"
+    page: Pages | None = None
+    action: Actions | ServerUpdate | None = None
+    server_id: int | None = None
+    image_id: int | None = None
     confirm: bool = False
-    image_id: int = 0
 
 
-class ServerList(CallbackData, prefix="server_list"):
-    action: str
-
-
-class ServerTypeSelect(CallbackData, prefix="server_type"):
-    server: int = 0
-    is_select: bool = False
-
-
-class LocationTypeSelect(CallbackData, prefix="location_type"):
-    location: int = 0
-    is_select: bool = False
-
-
-class ImageTypeSelect(CallbackData, prefix="image_type"):
-    image: int = 0
-    is_select: bool = False
+class SelectCB(CallbackData, prefix="select"):
+    key: str = "KEY"
+    page: Pages | None = None
+    action: Actions | None = None
+    datatype: ServerCreate | None = None
+    datavalue: str | int | None = None

keys/enums/__init__.py

@@ -0,0 +1,4 @@
+from .callbacks import Pages, Actions
+from .servers import ServerCreate, ServerUpdate
+
+__all__ = ["Pages", "Actions", "ServerCreate", "ServerUpdate"]

keys/enums/callbacks.py

@@ -0,0 +1,16 @@
+from enum import Enum
+
+
+class Actions(str, Enum):
+    HOME = "home"
+    LIST = "list"
+    INFO = "info"
+    CREATE = "create"
+    UPDATE = "update"
+    DELETE = "delete"
+
+
+class Pages(str, Enum):
+    HOME = "home"
+    MENU = "menu"
+    SERVER = "server"

keys/action.py → keys/enums/servers.py

@@ -1,14 +1,18 @@
 from enum import Enum
 
 
-class Actions(str, Enum):
+class ServerCreate(str, Enum):
+    LOCATION = "location"
+    SERVER = "server"
+    IMAGE = "image"
+
+
+class ServerUpdate(str, Enum):
     POWER_ON = "power_on"
     POWER_OFF = "power_off"
     REBOOT = "reboot"
     RESET_PASSWORD = "reset_password"
     DELETE = "delete"
-    HOME = "home"
-    INFO = "info"
     REBUILD = "rebuild"
-    UPDATE = "update"
     RESET = "reset"
+    UPDATE = "update"