mariadb -> postgres

app/core/background/convert_service.py

@@ -6,7 +6,7 @@ import json
 import shutil
 import magic  # python-magic for MIME detection
 from base58 import b58decode, b58encode
-from sqlalchemy import and_, or_
+from sqlalchemy import and_, or_, select
 from app.core.models.node_storage import StoredContent
 from app.core.models._telegram import Wrapped_CBotChat
 from app.core._utils.send_status import send_status
@@ -19,9 +19,9 @@ from app.core.content.content_id import ContentId
 
 
 async def convert_loop(memory):
-    with db_session() as session:
+    async with db_session() as session:
         # Query for unprocessed encrypted content
-        unprocessed_encrypted_content = session.query(StoredContent).filter(
+        unprocessed_encrypted_content = (await session.execute(select(StoredContent).where(
             and_(
                 StoredContent.type == "onchain/content",
                 or_(
@@ -29,15 +29,15 @@ async def convert_loop(memory):
                     StoredContent.ipfs_cid == None,
                 )
             )
-        ).first()
+        ))).scalars().first()
         if not unprocessed_encrypted_content:
             make_log("ConvertProcess", "No content to convert", level="debug")
             return
 
         # Достаем расшифрованный файл
-        decrypted_content = session.query(StoredContent).filter(
+        decrypted_content = (await session.execute(select(StoredContent).where(
             StoredContent.id == unprocessed_encrypted_content.decrypted_content_id
-        ).first()
+        ))).scalars().first()
         if not decrypted_content:
             make_log("ConvertProcess", "Decrypted content not found", level="error")
             return
@@ -78,7 +78,7 @@ async def convert_loop(memory):
                     option_name: decrypted_content.hash for option_name in ['high', 'low', 'low_preview']
                 }
             }
-            session.commit()
+            await session.commit()
             return
 
         # ==== Конвертация для видео или аудио: оригинальная логика ====
@@ -171,9 +171,7 @@ async def convert_loop(memory):
             file_hash = b58encode(bytes.fromhex(file_hash)).decode()
 
             # Save new StoredContent if not exists
-            if not session.query(StoredContent).filter(
-                StoredContent.hash == file_hash
-            ).first():
+            if not (await session.execute(select(StoredContent).where(StoredContent.hash == file_hash))).scalars().first():
                 new_content = StoredContent(
                     type="local/content_bin",
                     hash=file_hash,
@@ -183,7 +181,7 @@ async def convert_loop(memory):
                     created=datetime.now(),
                 )
                 session.add(new_content)
-                session.commit()
+                await session.commit()
 
             save_path = os.path.join(UPLOADS_DIR, file_hash)
             try:
@@ -233,13 +231,13 @@ async def convert_loop(memory):
             **unprocessed_encrypted_content.meta,
             'converted_content': converted_content
         }
-        session.commit()
+        await session.commit()
 
         # Notify user if needed
         if not unprocessed_encrypted_content.meta.get('upload_notify_msg_id'):
-            wallet_owner_connection = session.query(WalletConnection).filter(
+            wallet_owner_connection = (await session.execute(select(WalletConnection).where(
                 WalletConnection.wallet_address == unprocessed_encrypted_content.owner_address
-            ).order_by(WalletConnection.id.desc()).first()
+            ).order_by(WalletConnection.id.desc()))).scalars().first()
             if wallet_owner_connection:
                 wallet_owner_user = wallet_owner_connection.user
                 bot = Wrapped_CBotChat(
@@ -249,7 +247,7 @@ async def convert_loop(memory):
                     db_session=session
                 )
                 unprocessed_encrypted_content.meta['upload_notify_msg_id'] = await bot.send_content(session, unprocessed_encrypted_content)
-                session.commit()
+                await session.commit()
 
 
 async def main_fn(memory):

app/core/background/indexer_service.py

@@ -17,6 +17,7 @@ from app.core._utils.resolve_content import resolve_content
 from app.core.models.wallet_connection import WalletConnection
 from app.core._keyboards import get_inline_keyboard
 from app.core.models._telegram import Wrapped_CBotChat
+from sqlalchemy import select, and_, desc
 from app.core.storage import db_session
 import os
 import traceback
@@ -33,7 +34,7 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
             platform_found = True
 
     make_log("Indexer", "Service running", level="debug")
-    with db_session() as session:
+    async with db_session() as session:
         try:
             result = await toncenter.run_get_method('EQD8TJ8xEWB1SpnRE4d89YO3jl0W0EiBnNS4IBaHaUmdfizE', 'get_pool_data')
             assert result['exit_code'] == 0, f"Error in get-method: {result}"
@@ -41,21 +42,21 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
             assert result['stack'][1][0] == 'num', f"get second element is not num"
             usdt_per_ton = (int(result['stack'][0][1], 16) * 1e3) / int(result['stack'][1][1], 16)
             ton_per_star = 0.014 / usdt_per_ton
-            ServiceConfig(session).set('live_tonPerStar', [ton_per_star, datetime.utcnow().timestamp()])
+            await ServiceConfig(session).set('live_tonPerStar', [ton_per_star, datetime.utcnow().timestamp()])
             make_log("TON_Daemon", f"TON per STAR price: {ton_per_star}", level="DEBUG")
         except BaseException as e:
             make_log("TON_Daemon", f"Error while saving TON per STAR price: {e}" + '\n' + traceback.format_exc(), level="ERROR")
 
-        new_licenses = session.query(UserContent).filter(
+        new_licenses = (await session.execute(select(UserContent).where(
             and_(
                 ~UserContent.meta.contains({'notification_sent': True}),
                 UserContent.type == 'nft/listen'
             )
-        ).all()
+        ))).scalars().all()
         for new_license in new_licenses:
-            licensed_content = session.query(StoredContent).filter(
+            licensed_content = (await session.execute(select(StoredContent).where(
                 StoredContent.id == new_license.content_id
-            ).first()
+            ))).scalars().first()
             if not licensed_content:
                 make_log("Indexer", f"Licensed content not found: {new_license.content_id}", level="error")
 
@@ -70,10 +71,12 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
                             session, licensed_content
                         )
 
-                    wallet_owner_connection = session.query(WalletConnection).filter_by(
-                        wallet_address=licensed_content.owner_address,
-                        invalidated=False
-                    ).order_by(desc(WalletConnection.id)).first()
+                    wallet_owner_connection = (await session.execute(
+                        select(WalletConnection).where(
+                            WalletConnection.wallet_address == licensed_content.owner_address,
+                            WalletConnection.invalidated == False
+                        ).order_by(desc(WalletConnection.id))
+                    )).scalars().first()
                     wallet_owner_user = wallet_owner_connection.user
                     if wallet_owner_user.telegram_id:
                         wallet_owner_bot = Wrapped_CBotChat(memory._telegram_bot, chat_id=wallet_owner_user.telegram_id, user=wallet_owner_user, db_session=session)
@@ -89,21 +92,19 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
                     make_log("IndexerSendNewLicense", f"Error: {e}" + '\n' + traceback.format_exc(), level="error")
 
             new_license.meta = {**new_license.meta, 'notification_sent': True}
-            session.commit()
+            await session.commit()
 
-        content_without_cid = session.query(StoredContent).filter(
-            StoredContent.content_id == None
-        )
+        content_without_cid = (await session.execute(select(StoredContent).where(StoredContent.content_id == None))).scalars().all()
         for target_content in content_without_cid:
             target_cid = target_content.cid.serialize_v2()
             make_log("Indexer", f"Content without CID: {target_content.hash}, setting CID: {target_cid}", level="debug")
             target_content.content_id = target_cid
 
-        session.commit()
+        await session.commit()
 
-        last_known_index_ = session.query(StoredContent).filter(
-            StoredContent.onchain_index != None
-        ).order_by(StoredContent.onchain_index.desc()).first()
+        last_known_index_ = (await session.execute(
+            select(StoredContent).where(StoredContent.onchain_index != None).order_by(StoredContent.onchain_index.desc())
+        )).scalars().first()
         last_known_index = last_known_index_.onchain_index if last_known_index_ else 0
         last_known_index = max(last_known_index, 0)
         make_log("Indexer", f"Last known index: {last_known_index}", level="debug")
@@ -196,14 +197,13 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
 
         user_wallet_connection = None
         if item_owner_address:
-            user_wallet_connection = session.query(WalletConnection).filter(
+            user_wallet_connection = (await session.execute(select(WalletConnection).where(
                 WalletConnection.wallet_address == item_owner_address.to_string(1, 1, 1)
-            ).first()
+            ))).scalars().first()
 
-        encrypted_stored_content = session.query(StoredContent).filter(
-            StoredContent.hash == item_content_hash_str,
-            # StoredContent.type.like("local%")
-        ).first()
+        encrypted_stored_content = (await session.execute(select(StoredContent).where(
+            StoredContent.hash == item_content_hash_str
+        ))).scalars().first()
         if encrypted_stored_content:
             is_duplicate = encrypted_stored_content.type.startswith("onchain") \
                            and encrypted_stored_content.onchain_index != item_index
@@ -234,14 +234,15 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
                         )
 
                         try:
-                            for hint_message in session.query(KnownTelegramMessage).filter(
+                            result = await session.execute(select(KnownTelegramMessage).where(
                                 and_(
                                     KnownTelegramMessage.chat_id == user.telegram_id,
                                     KnownTelegramMessage.type == 'hint',
                                     cast(KnownTelegramMessage.meta['encrypted_content_hash'], String) == encrypted_stored_content.hash,
                                     KnownTelegramMessage.deleted == False
                                 )
-                            ).all():
+                            ))
+                            for hint_message in result.scalars().all():
                                 await user_uploader_wrapper.delete_message(hint_message.message_id)
                         except BaseException as e:
                             make_log("Indexer", f"Error while deleting hint messages: {e}" + '\n' + traceback.format_exc(), level="error")
@@ -260,7 +261,7 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
                     **item_metadata_packed
                 }
 
-                session.commit()
+                await session.commit()
                 return platform_found, seqno
             else:
                 item_metadata_packed['copied_from'] = encrypted_stored_content.id
@@ -282,7 +283,7 @@ async def indexer_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
             updated=datetime.now()
         )
         session.add(onchain_stored_content)
-        session.commit()
+        await session.commit()
         make_log("Indexer", f"Item indexed: {item_content_hash_str}", level="info")
         last_known_index += 1
 

app/core/background/license_service.py

@@ -3,7 +3,7 @@ from base64 import b64decode
 from datetime import datetime, timedelta
 
 from base58 import b58encode
-from sqlalchemy import and_, or_
+from sqlalchemy import and_, or_, select, desc
 from tonsdk.boc import Cell
 from tonsdk.utils import Address
 
@@ -27,7 +27,7 @@ import traceback
 
 async def license_index_loop(memory, platform_found: bool, seqno: int) -> [bool, int]:
     make_log("LicenseIndex", "Service running", level="debug")
-    with db_session() as session:
+    async with db_session() as session:
         async def check_telegram_stars_transactions():
             # Проверка звездных telegram транзакций, обновление paid
             offset = {'desc': 'Статичное число заранее известного количества транзакций, которое даже не знает наш бот', 'value': 1}['value'] + \
@@ -45,19 +45,19 @@ async def license_index_loop(memory, platform_found: bool, seqno: int) -> [bool,
                         continue
 
                     try:
-                        existing_invoice = session.query(StarsInvoice).filter(
+                        existing_invoice = (await session.execute(select(StarsInvoice).where(
                             StarsInvoice.external_id == star_payment.source.invoice_payload
-                        ).first()
+                        ))).scalars().first()
                         if not existing_invoice:
                             continue
 
                         if star_payment.amount == existing_invoice.amount:
                             if not existing_invoice.paid:
                                 existing_invoice.paid = True
-                                session.commit()
+                                await session.commit()
 
-                                licensed_content = session.query(StoredContent).filter(StoredContent.hash == existing_invoice.content_hash).first()
-                                user = session.query(User).filter(User.id == existing_invoice.user_id).first()
+                                licensed_content = (await session.execute(select(StoredContent).where(StoredContent.hash == existing_invoice.content_hash))).scalars().first()
+                                user = (await session.execute(select(User).where(User.id == existing_invoice.user_id))).scalars().first()
 
                                 await (Wrapped_CBotChat(memory._client_telegram_bot, chat_id=user.telegram_id, user=user, db_session=session)).send_content(
                                     session, licensed_content
@@ -73,9 +73,10 @@ async def license_index_loop(memory, platform_found: bool, seqno: int) -> [bool,
             make_log("StarsProcessing", f"Error: {e}" + '\n' + traceback.format_exc(), level="error")
 
         # Проверка кошельков пользователей на появление новых NFT, добавление их в базу как неопознанные
-        for user in session.query(User).filter(
+        users = (await session.execute(select(User).where(
             User.last_use > datetime.now() - timedelta(hours=4)
-        ).order_by(User.updated.asc()).all():
+        ).order_by(User.updated.asc()))).scalars().all()
+        for user in users:
             user_wallet_address = user.wallet_address(session)
             if not user_wallet_address:
                 make_log("LicenseIndex", f"User {user.id} has no wallet address", level="info")
@@ -91,17 +92,17 @@ async def license_index_loop(memory, platform_found: bool, seqno: int) -> [bool,
             try:
                 await user.scan_owned_user_content(session)
                 user.meta = {**user.meta, 'last_updated_licenses': datetime.now().isoformat()}
-                session.commit()
+                await session.commit()
             except BaseException as e:
                 make_log("LicenseIndex", f"Error: {e}" + '\n' + traceback.format_exc(), level="error")
         
         # Проверка NFT на актуальность данных, в том числе уже проверенные
-        process_content = session.query(UserContent).filter(
+        process_content = (await session.execute(select(UserContent).where(
             and_(
                 UserContent.type.startswith('nft/'),
                 UserContent.updated < (datetime.now() - timedelta(minutes=60)),
             )
-        ).order_by(UserContent.updated.asc()).first()
+        ).order_by(UserContent.updated.asc()))).scalars().first()
         if process_content:
             make_log("LicenseIndex", f"Syncing content with blockchain: {process_content.id}", level="info")
             try:
@@ -110,7 +111,7 @@ async def license_index_loop(memory, platform_found: bool, seqno: int) -> [bool,
                 make_log("LicenseIndex", f"Error: {e}" + '\n' + traceback.format_exc(), level="error")
             finally:
                 process_content.updated = datetime.now()
-                session.commit()
+                await session.commit()
 
     return platform_found, seqno
 

app/core/background/ton_service.py

@@ -125,7 +125,7 @@ async def main_fn(memory):
             sw_seqno_value = await get_sw_seqno()
             make_log("TON", f"Service running ({sw_seqno_value})", level="debug")
 
-            with db_session() as session:
+            async with db_session() as session:
                 # Проверка отправленных сообщений
                 await send_status("ton_daemon", f"working: processing in-txs (seqno={sw_seqno_value})")
                 async def process_incoming_transaction(transaction: dict):
@@ -142,14 +142,17 @@ async def main_fn(memory):
                         in_msg_created_at = in_msg_slice.read_uint(64)
                         in_msg_epoch = int(in_msg_created_at // (60 * 60))
                         in_msg_seqno = HighloadQueryId.from_query_id(in_msg_query_id).to_seqno()
+                        from sqlalchemy import select
                         in_msg_blockchain_task = (
-                            session.query(BlockchainTask).filter(
-                                and_(
-                                    BlockchainTask.seqno == in_msg_seqno,
-                                    BlockchainTask.epoch == in_msg_epoch,
+                            await session.execute(
+                                select(BlockchainTask).where(
+                                    and_(
+                                        BlockchainTask.seqno == in_msg_seqno,
+                                        BlockchainTask.epoch == in_msg_epoch,
+                                    )
                                 )
                             )
-                        ).first()
+                        ).scalars().first()
                         if not in_msg_blockchain_task:
                             return
 
@@ -157,7 +160,7 @@ async def main_fn(memory):
                             in_msg_blockchain_task.status = 'done'
                             in_msg_blockchain_task.transaction_hash = transaction_hash
                             in_msg_blockchain_task.transaction_lt = transaction_lt
-                            session.commit()
+                            await session.commit()
 
                     for blockchain_message in [transaction['in_msg']]:
                         try:
@@ -177,11 +180,11 @@ async def main_fn(memory):
 
                 await send_status("ton_daemon", f"working: processing out-txs (seqno={sw_seqno_value})")
                 # Отправка подписанных сообщений
-                for blockchain_task in (
-                    session.query(BlockchainTask).filter(
-                        BlockchainTask.status == 'processing',
-                    ).order_by(BlockchainTask.updated.asc()).all()
-                ):
+                from sqlalchemy import select
+                _processing = (await session.execute(select(BlockchainTask).where(
+                    BlockchainTask.status == 'processing'
+                ).order_by(BlockchainTask.updated.asc()))).scalars().all()
+                for blockchain_task in _processing:
                     make_log("TON_Daemon", f"Processing task (processing) {blockchain_task.id}")
                     query_boc = bytes.fromhex(blockchain_task.meta['signed_message'])
                     errors_list = []
@@ -210,23 +213,22 @@ async def main_fn(memory):
                             # or sum([int("terminating vm with exit code 36" in e) for e in errors_list]) > 0:
                         make_log("TON_Daemon", f"Task {blockchain_task.id} done", level="DEBUG")
                         blockchain_task.status = 'done'
-                        session.commit()
+                        await session.commit()
                         continue
 
                     await asyncio.sleep(0.5)
 
                 await send_status("ton_daemon", f"working: creating new messages (seqno={sw_seqno_value})")
                 # Создание новых подписей
-                for blockchain_task in (
-                    session.query(BlockchainTask).filter(BlockchainTask.status == 'wait').all()
-                ):
+                _waiting = (await session.execute(select(BlockchainTask).where(BlockchainTask.status == 'wait'))).scalars().all()
+                for blockchain_task in _waiting:
                     try:
                         # Check processing tasks in current epoch < 3_000_000
-                        if (
-                            session.query(BlockchainTask).filter(
-                                BlockchainTask.epoch == blockchain_task.epoch,
-                            ).count() > 3_000_000
-                        ):
+                        from sqlalchemy import func
+                        _cnt = (await session.execute(select(func.count()).select_from(BlockchainTask).where(
+                            BlockchainTask.epoch == blockchain_task.epoch
+                        ))).scalar() or 0
+                        if _cnt > 3_000_000:
                             make_log("TON", f"Too many processing tasks in epoch {blockchain_task.epoch}", level="error")
                             await send_status("ton_daemon", f"working: too many tasks in epoch {blockchain_task.epoch}")
                             await asyncio.sleep(5)
@@ -235,10 +237,11 @@ async def main_fn(memory):
                         sign_created = int(datetime.utcnow().timestamp()) - 60
                         try:
                             current_epoch = int(datetime.utcnow().timestamp() // (60 * 60))
+                            from sqlalchemy import func
                             max_epoch_seqno = (
-                                session.query(func.max(BlockchainTask.seqno)).filter(
+                                (await session.execute(select(func.max(BlockchainTask.seqno)).where(
                                     BlockchainTask.epoch == current_epoch
-                                ).scalar() or 0
+                                ))).scalar() or 0
                             )
                             current_epoch_shift = 3_000_000 if current_epoch % 2 == 0 else 0
                             current_seqno = max_epoch_seqno + 1 + (current_epoch_shift if max_epoch_seqno == 0 else 0)
@@ -266,7 +269,7 @@ async def main_fn(memory):
                             'sign_created': sign_created,
                             'signed_message': query_boc.hex(),
                         }
-                        session.commit()
+                        await session.commit()
                         make_log("TON", f"Created signed message for task {blockchain_task.id}" + '\n' + traceback.format_exc(), level="info")
                     except BaseException as e:
                         make_log("TON", f"Error processing task {blockchain_task.id}: {e}" + '\n' + traceback.format_exc(), level="error")
@@ -287,4 +290,3 @@ async def main_fn(memory):
 
 
 
-