Skip to content

Instantly share code, notes, and snippets.

@gabbhack
Last active December 3, 2022 09:28
Show Gist options
  • Save gabbhack/166699945458cbd90a49bed8af95fbc8 to your computer and use it in GitHub Desktop.
Save gabbhack/166699945458cbd90a49bed8af95fbc8 to your computer and use it in GitHub Desktop.
from gino import Gino
db = Gino()
async def start():
await db.set_bind('postgresql://localhost/gino')
# Create tables
await db.gino.create_all()
import database
# you must import all models for them to appear in the metadata
from models import *
# blabla
async def on_startup(_):
await database.start()
start_webhook(..., on_startup=on_startup)
from aiogram.dispatcher.middlewares import BaseMiddleware
from aiogram import types
from models import User
class MetricMiddleware(BaseMiddleware):
async def on_process_message(self, message: types.Message, data: dict):
user = await User.get_or_create(message.from_user.id)
await user.up_messages()
data["user"] = user
from aiogram import types
from models import User
@dp.message_handler(commands=['count'])
async def test(message: types.Message, user: User):
await message.answer(f"The bot has processed {user.messages} messages from you.")
# As planned, this file is located in the "models" directory.
from database import db
class User(db.Model):
__tablename__ = 'users'
id: int = db.Column(db.Integer(), primary_key=True)
messages: int = db.Column(db.Integer(), default=0)
@classmethod
async def get_or_create(cls, id) -> "User":
user = await cls.get(id)
if user is None:
return await cls.create(id=id)
return user
async def up_messages(self) -> int:
"""
Update the statistics of the messages.
Return the updated value.
"""
await self.update(messages=self.messages+1).apply()
return self.messages
Sign up for free to join this conversation on GitHub. Already have an account? Sign in to comment