Initial MCM project: FastAPI + Textual TUI unified messenger
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. Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
73
services/contact_service.py
Normal file
73
services/contact_service.py
Normal file
@@ -0,0 +1,73 @@
|
||||
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
|
||||
Reference in New Issue
Block a user