#!/usr/bin/env python
from datetime import datetime
from sqlalchemy import ForeignKey
from sqlalchemy import create_engine, delete, select, update
from sqlalchemy.orm import DeclarativeBase, Mapped
from sqlalchemy.orm import mapped_column, relationship, sessionmaker
class Base(DeclarativeBase):
pass
class User(Base):
__tablename__ = "user"
id: Mapped[int] = mapped_column(primary_key=True)
address_first_id: Mapped[int] = mapped_column(ForeignKey('address.id'))
address_first: Mapped["Address"] = relationship(foreign_keys=[address_first_id])
address_second_id: Mapped[int] = mapped_column(ForeignKey('address.id'))
address_second: Mapped["Address"] = relationship(foreign_keys=[address_second_id])
is_available: Mapped[bool] = mapped_column(default=True)
created_on: Mapped[datetime] = mapped_column(default=datetime.utcnow())
updated_on: Mapped[datetime] = mapped_column(default=datetime.utcnow(), onupdate=datetime.utcnow())
class Address(Base):
__tablename__ = "address"
id: Mapped[int] = mapped_column(primary_key=True)
comment: Mapped[str] = mapped_column(unique=True, nullable=False)
is_available: Mapped[bool] = mapped_column(default=True)
created_on: Mapped[datetime] = mapped_column(default=datetime.utcnow())
updated_on: Mapped[datetime] = mapped_column(default=datetime.utcnow(), onupdate=datetime.utcnow())
if __name__ == "__main__":
engine = create_engine('sqlite:///:memory:', echo=True)
Base.metadata.create_all(engine)
Session = sessionmaker(engine)
# добавим пользователя с двумя адресами
with Session.begin() as session:
address1 = Address(comment="Lenina 45")
address2 = Address(comment="Sedina 5")
user = User(address_first=address1, address_second=address2)
session.add(user)
# проверим, что пользователь был создан успешно
with Session.begin() as session:
user = session.query(User).first()
print(user, user.address_first.comment)
with Session.begin() as session:
# находим первый адрес по части адреса
adr = session.query(Address).where(Address.comment.like("%Sedina%")).first()
print(adr, adr.comment)
# находим первого пользователя у которого один из адресов соответствует найденному выше адресу
usr = session.query(User).where(User.address_first == adr or User.address_second == adr).first()
print(usr, usr.address_first.comment, usr.address_second.comment)
def cmd_restart_service(update, context):
if update.message.from_user.id == XXX:
import subprocess
update.message.reply_text("Будет исполнено...")
subprocess.run("systemctl --user restart mybot.service".split())
else:
update.message.reply_text("Ай! Не делай так больше...")
from time import sleep
sleep(1)
...
TypeError: Object of type Tag is not JSON serializable
>>> bool(0)
False
>>> bool(not 0)
True
if a and a_2 and b and b_2 and c and c_2 or a and a_2 and not b and not b_2 and c and c_2 or not a and not a_2 and b and b_2 and c and c_2
>>> for item in zip([1, 2, 3], ['sugar', 'spice', 'everything nice']):
... print(item)
...
(1, 'sugar')
(2, 'spice')
(3, 'everything nice')
items = program.find("div", class_="p-programms").strings
joined_items = "\n".join(items)
print(f"{section} | {joined_items}")
for program in program:
def cmd_about_movie(update, context):
# функция обработчик
# может например обрабатывать имя команды, как идентификатор для вывода контента
pass
updater = Updater(TOKEN, use_context=True)
updater.dispatcher.add_handler(CommandHandler("b739423", cmd_about_movie))
def cmd_restart_service(update, context):
if update.message.from_user.id == 777777777: # id пользователя
import subprocess
update.message.reply_text("будет исполнено...")
subprocess.run("sudo supervisorctl restart mybotname".split())
l = [1, 0.6, 0.8, 0.3]
l_len = len(l)
res = [[l[abs(x-y)] for y in range(l_len)] for x in range(l_len)]
print(res) # [[1, 0.6, 0.8, 0.3], [0.6, 1, 0.6, 0.8], [0.8, 0.6, 1, 0.6], [0.3, 0.8, 0.6, 1]]