80 lines
2.4 KiB
Python
80 lines
2.4 KiB
Python
# SQL
|
|
from sqlmodel import Field, SQLModel, or_, select
|
|
from sqlmodel.ext.asyncio.session import AsyncSession
|
|
from sqlalchemy.ext.asyncio import create_async_engine
|
|
# CONFIGS
|
|
from os import path
|
|
import json
|
|
import aiofiles
|
|
|
|
from datetime import datetime
|
|
|
|
#------------------------------------------------------------
|
|
# INIT
|
|
#------------------------------------------------------------
|
|
global engine
|
|
def init_engine():
|
|
global engine
|
|
sqlite_file_name = "database.db"
|
|
sqlite_url = f"sqlite+aiosqlite:///{sqlite_file_name}"
|
|
engine = create_async_engine(sqlite_url)
|
|
|
|
async def create_db_and_tables():
|
|
async with engine.begin() as conn:
|
|
await conn.run_sync(SQLModel.metadata.create_all)
|
|
|
|
async def get_session():
|
|
async with AsyncSession(engine) as session:
|
|
yield session
|
|
|
|
#------------------------- END ------------------------------
|
|
|
|
|
|
#------------------------------------------------------------
|
|
# TABLES
|
|
#------------------------------------------------------------
|
|
class UserToken(SQLModel, table=True):
|
|
def __init__(self, token, logs=""):
|
|
self.token = token
|
|
self.logs = logs
|
|
self.issue_date = datetime.today().strftime('%Y-%m-%d')
|
|
token: str = Field(max_length=200, primary_key=True)
|
|
logs: str = Field()
|
|
issue_date: str = Field()
|
|
#------------------------- END ------------------------------
|
|
|
|
|
|
#------------------------------------------------------------
|
|
# JSONS / CONFIGS
|
|
#------------------------------------------------------------
|
|
config_file = 'config.json'
|
|
if not path.exists(config_file):
|
|
db = {'system_api_token': 'None',
|
|
'system_api_url': 'None'}
|
|
js = json.dumps(db, indent=2)
|
|
with open(config_file, "w") as outfile:
|
|
outfile.write(js)
|
|
print(f'Created new {config_file}')
|
|
|
|
async def read(file=config_file):
|
|
async with aiofiles.open(file, "r", encoding="utf-8") as openfile:
|
|
content = await openfile.read()
|
|
db = json.loads(content)
|
|
return db
|
|
|
|
async def write(db, file=config_file):
|
|
js = json.dumps(db, indent=2, ensure_ascii=False)
|
|
async with aiofiles.open(file, "w", encoding="utf-8") as outfile:
|
|
await outfile.write(js)
|
|
|
|
#------------------------- END ------------------------------
|
|
|
|
async def user_token_in_db_func(
|
|
session: AsyncSession,
|
|
user_token: str
|
|
):
|
|
statement = select(UserToken).where(UserToken.token == user_token)
|
|
result = await session.exec(statement)
|
|
# Get the first result or None if not found
|
|
user_token_obj = result.first()
|
|
return user_token_obj |