itdrui.de
7f3b4768c3
MultiCustomerMessenger supporting Telegram (python-telegram-bot), WhatsApp (Green API) and SMS (python-gsmmodem-new). REST API with Bearer-token auth, SQLAlchemy models for MariaDB, APScheduler for background polling, and Textual TUI running in same asyncio event-loop.
74 lines
2.0 KiB
Python
74 lines
2.0 KiB
Python
from __future__ import annotations
|
|
|
|
from sqlalchemy.orm import Session
|
|
|
|
from db.models import Contact
|
|
from schemas import ContactCreate, ContactUpdate
|
|
|
|
|
|
def get_all(db: Session) -> list[Contact]:
|
|
return db.query(Contact).order_by(Contact.name).all()
|
|
|
|
|
|
def get_by_id(db: Session, contact_id: str) -> Contact | None:
|
|
return db.query(Contact).filter(Contact.id == contact_id).first()
|
|
|
|
|
|
def get_by_telegram_id(db: Session, telegram_id: str) -> Contact | None:
|
|
return db.query(Contact).filter(Contact.telegram_id == telegram_id).first()
|
|
|
|
|
|
def get_by_phone(db: Session, phone: str) -> Contact | None:
|
|
return (
|
|
db.query(Contact)
|
|
.filter((Contact.phone == phone) | (Contact.whatsapp_phone == phone))
|
|
.first()
|
|
)
|
|
|
|
|
|
def create(db: Session, data: ContactCreate) -> Contact:
|
|
contact = Contact(**data.model_dump(exclude_none=False))
|
|
db.add(contact)
|
|
db.commit()
|
|
db.refresh(contact)
|
|
return contact
|
|
|
|
|
|
def update(db: Session, contact: Contact, data: ContactUpdate) -> Contact:
|
|
for field, value in data.model_dump(exclude_unset=True).items():
|
|
setattr(contact, field, value)
|
|
db.commit()
|
|
db.refresh(contact)
|
|
return contact
|
|
|
|
|
|
def delete(db: Session, contact: Contact) -> None:
|
|
db.delete(contact)
|
|
db.commit()
|
|
|
|
|
|
def get_or_create_by_telegram(
|
|
db: Session, telegram_id: str, name: str, username: str | None = None
|
|
) -> Contact:
|
|
contact = get_by_telegram_id(db, telegram_id)
|
|
if not contact:
|
|
contact = create(
|
|
db,
|
|
ContactCreate(
|
|
name=name,
|
|
telegram_id=telegram_id,
|
|
telegram_username=username,
|
|
),
|
|
)
|
|
return contact
|
|
|
|
|
|
def get_or_create_by_phone(db: Session, phone: str, name: str | None = None) -> Contact:
|
|
contact = get_by_phone(db, phone)
|
|
if not contact:
|
|
contact = create(
|
|
db,
|
|
ContactCreate(name=name or phone, phone=phone, whatsapp_phone=phone),
|
|
)
|
|
return contact
|