89 lines
3.4 KiB
Python
89 lines
3.4 KiB
Python
from core.models.Model import Model
|
|
from core.models.session.SessionConnection import SessionConnection
|
|
from core.models.system.SystemConnection import SystemConnection
|
|
from dataclasses import dataclass
|
|
from typing import Union, Optional
|
|
|
|
_table_name: str = 'subscription_plans'
|
|
|
|
_table_definition: str = """
|
|
'id' int UNIQUE,
|
|
'code' varchar UNIQUE,
|
|
'wireguard_session_limit' int,
|
|
'duration' int,
|
|
'price' int,
|
|
'features_proxy' bool,
|
|
'features_wireguard' bool
|
|
"""
|
|
|
|
|
|
@dataclass
|
|
class SubscriptionPlan(Model):
|
|
id: int
|
|
code: str
|
|
wireguard_session_limit: int
|
|
duration: int
|
|
price: int
|
|
features_proxy: bool
|
|
features_wireguard: bool
|
|
|
|
@staticmethod
|
|
def find_by_id(id: int):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_one('SELECT * FROM subscription_plans WHERE id = ? LIMIT 1', SubscriptionPlan.factory, [id])
|
|
|
|
@staticmethod
|
|
def find(connection: Union[SessionConnection, SystemConnection], duration: int):
|
|
|
|
features_proxy = False
|
|
features_wireguard = False
|
|
|
|
if type(connection).__name__ == 'SessionConnection':
|
|
if connection.masked:
|
|
features_proxy = True
|
|
|
|
if connection.code == 'wireguard':
|
|
features_wireguard = True
|
|
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
return Model._query_one('SELECT * FROM subscription_plans WHERE features_proxy = ? AND features_wireguard = ? AND duration = ? LIMIT 1', SubscriptionPlan.factory, [features_proxy, features_wireguard, duration])
|
|
|
|
@staticmethod
|
|
def all(connection: Optional[Union[SessionConnection, SystemConnection]] = None):
|
|
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
|
|
if connection is None:
|
|
return Model._query_all('SELECT * FROM subscription_plans', SubscriptionPlan.factory)
|
|
|
|
else:
|
|
|
|
features_proxy = False
|
|
features_wireguard = False
|
|
|
|
if type(connection).__name__ == 'SessionConnection':
|
|
if connection.masked:
|
|
features_proxy = True
|
|
|
|
if connection.code == 'wireguard':
|
|
features_wireguard = True
|
|
|
|
return Model._query_all('SELECT * FROM subscription_plans WHERE features_proxy = ? AND features_wireguard = ?', SubscriptionPlan.factory, [features_proxy, features_wireguard])
|
|
|
|
@staticmethod
|
|
def truncate():
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition, drop_existing=True)
|
|
|
|
@staticmethod
|
|
def save_many(subscription_plans):
|
|
Model._create_table_if_not_exists(table_name=_table_name, table_definition=_table_definition)
|
|
Model._insert_many('INSERT INTO subscription_plans VALUES(?, ?, ?, ?, ?, ?, ?)', SubscriptionPlan.tuple_factory, subscription_plans)
|
|
|
|
@staticmethod
|
|
def factory(cursor, row):
|
|
local_fields = [column[0] for column in cursor.description]
|
|
return SubscriptionPlan(**{key: value for key, value in zip(local_fields, row)})
|
|
|
|
@staticmethod
|
|
def tuple_factory(subscription_plan):
|
|
return subscription_plan.id, subscription_plan.code, subscription_plan.wireguard_session_limit, subscription_plan.duration, subscription_plan.price, subscription_plan.features_proxy, subscription_plan.features_wireguard
|