itdrui.de
b0c6ba44de
- _render_message: leeren style-Tag vermieden (MarkupError bei inbound-Msgs) - Nachrichtentext: eckige Klammern werden escaped (kein Markup-Injection) - get_by_phone: sucht +49xxx und 49xxx gleichzeitig (Green API liefert ohne +)
77 lines
2.1 KiB
Python
77 lines
2.1 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:
|
|
# Normalisierung: mit und ohne führendes + suchen
|
|
variants = {phone, "+" + phone.lstrip("+"), phone.lstrip("+")}
|
|
from sqlalchemy import or_
|
|
conditions = or_(*(
|
|
(Contact.phone == v) | (Contact.whatsapp_phone == v)
|
|
for v in variants
|
|
))
|
|
return db.query(Contact).filter(conditions).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
|