admin.schubert/Olympus
1
0
Fork 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

feat(customers): add dashboard and customer management

Browse source
This commit is contained in:
Schubert Ferenc 2026-07-02 23:37:46 +02:00
parent 694b7bd09a
commit 92cb8d1286
28 changed files with 2521 additions and 13 deletions
Download patch file Download diff file Expand all files Collapse all files

94
backend/hermes/alembic/versions/9c1d8a2f6b44_create_customers.py Normal file
View file

@ -0,0 +1,94 @@
"""create customers
Revision ID: 9c1d8a2f6b44
Revises: 7b2c9f4a0d31
Create Date: 2026-07-02 14:00:00.000000
"""
from typing import Sequence, Union
from alembic import op
import sqlalchemy as sa
revision: str = "9c1d8a2f6b44"
down_revision: Union[str, Sequence[str], None] = "7b2c9f4a0d31"
branch_labels: Union[str, Sequence[str], None] = None
depends_on: Union[str, Sequence[str], None] = None
def upgrade() -> None:
op.create_table(
"customers",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("customer_number", sa.String(length=50), nullable=False),
sa.Column("company_name", sa.String(length=255), nullable=False),
sa.Column("legal_name", sa.String(length=255), server_default="", nullable=False),
sa.Column("customer_type", sa.String(length=50), nullable=False),
sa.Column("status", sa.String(length=50), nullable=False),
sa.Column("industry", sa.String(length=120), server_default="", nullable=False),
sa.Column("website", sa.String(length=255), server_default="", nullable=False),
sa.Column("email", sa.String(length=255), server_default="", nullable=False),
sa.Column("phone", sa.String(length=80), server_default="", nullable=False),
sa.Column("tax_number", sa.String(length=120), server_default="", nullable=False),
sa.Column("vat_id", sa.String(length=120), server_default="", nullable=False),
sa.Column("notes", sa.Text(), server_default="", nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(op.f("ix_customers_company_name"), "customers", ["company_name"], unique=False)
op.create_index(op.f("ix_customers_customer_number"), "customers", ["customer_number"], unique=True)
op.create_index(op.f("ix_customers_customer_type"), "customers", ["customer_type"], unique=False)
op.create_index(op.f("ix_customers_status"), "customers", ["status"], unique=False)
op.create_table(
"customer_addresses",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("customer_id", sa.Integer(), nullable=False),
sa.Column("type", sa.String(length=50), nullable=False),
sa.Column("street", sa.String(length=255), server_default="", nullable=False),
sa.Column("postal_code", sa.String(length=30), server_default="", nullable=False),
sa.Column("city", sa.String(length=120), server_default="", nullable=False),
sa.Column("state", sa.String(length=120), server_default="", nullable=False),
sa.Column("country", sa.String(length=120), server_default="Deutschland", nullable=False),
sa.Column("is_primary", sa.Boolean(), server_default=sa.false(), nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.ForeignKeyConstraint(["customer_id"], ["customers.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(op.f("ix_customer_addresses_customer_id"), "customer_addresses", ["customer_id"], unique=False)
op.create_index(op.f("ix_customer_addresses_type"), "customer_addresses", ["type"], unique=False)
op.create_table(
"customer_contacts",
sa.Column("id", sa.Integer(), nullable=False),
sa.Column("customer_id", sa.Integer(), nullable=False),
sa.Column("first_name", sa.String(length=100), server_default="", nullable=False),
sa.Column("last_name", sa.String(length=100), server_default="", nullable=False),
sa.Column("position", sa.String(length=120), server_default="", nullable=False),
sa.Column("email", sa.String(length=255), server_default="", nullable=False),
sa.Column("phone", sa.String(length=80), server_default="", nullable=False),
sa.Column("mobile", sa.String(length=80), server_default="", nullable=False),
sa.Column("is_primary", sa.Boolean(), server_default=sa.false(), nullable=False),
sa.Column("notes", sa.Text(), server_default="", nullable=False),
sa.Column("created_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.Column("updated_at", sa.DateTime(timezone=True), server_default=sa.func.now(), nullable=False),
sa.ForeignKeyConstraint(["customer_id"], ["customers.id"], ondelete="CASCADE"),
sa.PrimaryKeyConstraint("id"),
)
op.create_index(op.f("ix_customer_contacts_customer_id"), "customer_contacts", ["customer_id"], unique=False)
def downgrade() -> None:
op.drop_index(op.f("ix_customer_contacts_customer_id"), table_name="customer_contacts")
op.drop_table("customer_contacts")
op.drop_index(op.f("ix_customer_addresses_type"), table_name="customer_addresses")
op.drop_index(op.f("ix_customer_addresses_customer_id"), table_name="customer_addresses")
op.drop_table("customer_addresses")
op.drop_index(op.f("ix_customers_status"), table_name="customers")
op.drop_index(op.f("ix_customers_customer_type"), table_name="customers")
op.drop_index(op.f("ix_customers_customer_number"), table_name="customers")
op.drop_index(op.f("ix_customers_company_name"), table_name="customers")
op.drop_table("customers")

208
backend/hermes/app/api/customers.py Normal file
View file

@ -0,0 +1,208 @@
import logging
from fastapi import APIRouter, Depends, HTTPException, Response, status
from sqlalchemy.orm import Session
from app.core.rbac import require_permission
from app.db.database import get_db
from app.models.customer import Customer, CustomerContact
from app.models.user import User
from app.repositories.customer_repository import CustomerRepository
from app.schemas.customer import (
CustomerContactCreate,
CustomerContactResponse,
CustomerContactUpdate,
CustomerCreate,
CustomerResponse,
CustomerUpdate,
)
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/customers",
tags=["Customers"],
)
def get_customer_or_404(db: Session, customer_id: int) -> Customer:
customer = CustomerRepository.get_by_id(db, customer_id)
if customer is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Kunde nicht gefunden",
)
return customer
def get_contact_or_404(db: Session, customer_id: int, contact_id: int) -> CustomerContact:
contact = CustomerRepository.get_contact_by_id(db, customer_id, contact_id)
if contact is None:
raise HTTPException(
status_code=status.HTTP_404_NOT_FOUND,
detail="Ansprechpartner nicht gefunden",
)
return contact
def ensure_unique_customer_number(
db: Session,
customer_number: str,
customer_id: int | None = None,
) -> None:
conflict = CustomerRepository.find_number_conflict(
db,
customer_number=customer_number,
exclude_customer_id=customer_id,
)
if conflict is not None:
raise HTTPException(
status_code=status.HTTP_409_CONFLICT,
detail="Kundennummer ist bereits vergeben",
)
@router.get("", response_model=list[CustomerResponse])
@router.get("/", response_model=list[CustomerResponse], include_in_schema=False)
def get_customers(
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("customers.read")),
):
logger.info("customers.list", extra={"actor_user_id": current_user.id})
return CustomerRepository.get_all(db)
@router.get("/{customer_id}", response_model=CustomerResponse)
def get_customer(
customer_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("customers.read")),
):
logger.info(
"customers.detail",
extra={"actor_user_id": current_user.id, "target_customer_id": customer_id},
)
return get_customer_or_404(db, customer_id)
@router.post("", response_model=CustomerResponse, status_code=status.HTTP_201_CREATED)
@router.post(
"/",
response_model=CustomerResponse,
status_code=status.HTTP_201_CREATED,
include_in_schema=False,
)
def create_customer(
customer: CustomerCreate,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("customers.create")),
):
ensure_unique_customer_number(db, customer.customer_number)
logger.info("customers.create", extra={"actor_user_id": current_user.id})
return CustomerRepository.create(db, customer)
@router.put("/{customer_id}", response_model=CustomerResponse)
def update_customer(
customer_id: int,
customer: CustomerUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("customers.update")),
):
db_customer = get_customer_or_404(db, customer_id)
ensure_unique_customer_number(db, customer.customer_number, customer_id)
logger.info(
"customers.update",
extra={"actor_user_id": current_user.id, "target_customer_id": customer_id},
)
return CustomerRepository.update(db, db_customer, customer)
@router.delete("/{customer_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_customer(
customer_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("customers.delete")),
):
db_customer = get_customer_or_404(db, customer_id)
logger.info(
"customers.delete",
extra={"actor_user_id": current_user.id, "target_customer_id": customer_id},
)
CustomerRepository.delete(db, db_customer)
return Response(status_code=status.HTTP_204_NO_CONTENT)
@router.get("/{customer_id}/contacts", response_model=list[CustomerContactResponse])
def get_customer_contacts(
customer_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("customers.read")),
):
get_customer_or_404(db, customer_id)
logger.info(
"customers.contacts.list",
extra={"actor_user_id": current_user.id, "target_customer_id": customer_id},
)
return CustomerRepository.get_contacts(db, customer_id)
@router.post(
"/{customer_id}/contacts",
response_model=CustomerContactResponse,
status_code=status.HTTP_201_CREATED,
)
def create_customer_contact(
customer_id: int,
contact: CustomerContactCreate,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("customers.update")),
):
get_customer_or_404(db, customer_id)
logger.info(
"customers.contacts.create",
extra={"actor_user_id": current_user.id, "target_customer_id": customer_id},
)
return CustomerRepository.create_contact(db, customer_id, contact)
@router.put("/{customer_id}/contacts/{contact_id}", response_model=CustomerContactResponse)
def update_customer_contact(
customer_id: int,
contact_id: int,
contact: CustomerContactUpdate,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("customers.update")),
):
get_customer_or_404(db, customer_id)
db_contact = get_contact_or_404(db, customer_id, contact_id)
logger.info(
"customers.contacts.update",
extra={
"actor_user_id": current_user.id,
"target_customer_id": customer_id,
"target_contact_id": contact_id,
},
)
return CustomerRepository.update_contact(db, db_contact, contact)
@router.delete("/{customer_id}/contacts/{contact_id}", status_code=status.HTTP_204_NO_CONTENT)
def delete_customer_contact(
customer_id: int,
contact_id: int,
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("customers.update")),
):
get_customer_or_404(db, customer_id)
db_contact = get_contact_or_404(db, customer_id, contact_id)
logger.info(
"customers.contacts.delete",
extra={
"actor_user_id": current_user.id,
"target_customer_id": customer_id,
"target_contact_id": contact_id,
},
)
CustomerRepository.delete_contact(db, db_contact)
return Response(status_code=status.HTTP_204_NO_CONTENT)

80
backend/hermes/app/api/dashboard.py Normal file
View file

@ -0,0 +1,80 @@
import logging
from fastapi import APIRouter, Depends
from sqlalchemy import func, select
from sqlalchemy.orm import Session
from app.core.rbac import get_user_permission_names, require_permission
from app.db.database import get_db
from app.models.rbac import Role
from app.models.user import User
from app.repositories.customer_repository import CustomerRepository
from app.repositories.user_repository import UserRepository
from app.schemas.dashboard import DashboardSummary, EmptyWidget, MetricCard
logger = logging.getLogger(__name__)
router = APIRouter(
prefix="/dashboard",
tags=["Dashboard"],
)
@router.get("/summary", response_model=DashboardSummary)
def get_dashboard_summary(
db: Session = Depends(get_db),
current_user: User = Depends(require_permission("dashboard.read")),
):
permissions = get_user_permission_names(current_user)
customers: list[MetricCard] = []
latest_customers = []
users: list[MetricCard] = []
roles: list[MetricCard] = []
if "customers.read" in permissions:
customers = [
MetricCard(label="Kunden", value=CustomerRepository.count_all(db)),
MetricCard(label="Aktive Kunden", value=CustomerRepository.count_active(db)),
MetricCard(label="Neue Kunden", value=CustomerRepository.count_recent(db)),
]
latest_customers = CustomerRepository.get_latest(db)
if "users.read" in permissions:
all_users = UserRepository.get_all(db)
users = [
MetricCard(label="Benutzer", value=len(all_users)),
MetricCard(label="Aktive Benutzer", value=len([user for user in all_users if user.is_active])),
]
if "roles.read" in permissions:
roles = [
MetricCard(
label="Rollen",
value=db.scalar(select(func.count(Role.id))) or 0,
)
]
logger.info("dashboard.summary", extra={"actor_user_id": current_user.id})
return DashboardSummary(
customers=customers,
latest_customers=latest_customers,
users=users,
roles=roles,
activities=EmptyWidget(
title="Letzte Aktivitäten",
message="Noch keine Aktivitäten vorhanden.",
),
tasks=EmptyWidget(
title="Offene Aufgaben",
message="Noch keine Aufgaben vorhanden.",
),
tickets=EmptyWidget(
title="Offene Tickets",
message="Noch keine Tickets vorhanden.",
),
projects=EmptyWidget(
title="Projekte",
message="Noch keine Projektdaten vorhanden.",
),
)

1
backend/hermes/app/db/database.py
View file

@ -21,6 +21,7 @@ class Base(DeclarativeBase):
# <<< HIER IMPORTIEREN >>>
import app.models.rbac
import app.models.customer
import app.models.user

4
backend/hermes/app/main.py
View file

@ -9,6 +9,8 @@ from starlette import status
from starlette.requests import Request
from app.api.auth import router as auth_router
from app.api.customers import router as customers_router
from app.api.dashboard import router as dashboard_router
from app.api.permissions import router as permissions_router
from app.api.roles import router as roles_router
from app.api.users import router as users_router
@ -26,6 +28,8 @@ app.include_router(auth_router)
app.include_router(users_router)
app.include_router(roles_router)
app.include_router(permissions_router)
app.include_router(customers_router)
app.include_router(dashboard_router)
logger = logging.getLogger(__name__)

92
backend/hermes/app/models/customer.py Normal file
View file

@ -0,0 +1,92 @@
from datetime import datetime
from sqlalchemy import Boolean, DateTime, ForeignKey, String, Text, func
from sqlalchemy.orm import Mapped, mapped_column, relationship
from app.db.database import Base
class Customer(Base):
__tablename__ = "customers"
id: Mapped[int] = mapped_column(primary_key=True)
customer_number: Mapped[str] = mapped_column(String(50), unique=True, index=True)
company_name: Mapped[str] = mapped_column(String(255), index=True)
legal_name: Mapped[str] = mapped_column(String(255), default="", server_default="")
customer_type: Mapped[str] = mapped_column(String(50), index=True)
status: Mapped[str] = mapped_column(String(50), index=True)
industry: Mapped[str] = mapped_column(String(120), default="", server_default="")
website: Mapped[str] = mapped_column(String(255), default="", server_default="")
email: Mapped[str] = mapped_column(String(255), default="", server_default="")
phone: Mapped[str] = mapped_column(String(80), default="", server_default="")
tax_number: Mapped[str] = mapped_column(String(120), default="", server_default="")
vat_id: Mapped[str] = mapped_column(String(120), default="", server_default="")
notes: Mapped[str] = mapped_column(Text, default="", server_default="")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
)
addresses: Mapped[list["CustomerAddress"]] = relationship(
back_populates="customer",
cascade="all, delete-orphan",
lazy="selectin",
)
contacts: Mapped[list["CustomerContact"]] = relationship(
back_populates="customer",
cascade="all, delete-orphan",
lazy="selectin",
)
class CustomerAddress(Base):
__tablename__ = "customer_addresses"
id: Mapped[int] = mapped_column(primary_key=True)
customer_id: Mapped[int] = mapped_column(
ForeignKey("customers.id", ondelete="CASCADE"),
index=True,
)
type: Mapped[str] = mapped_column(String(50), index=True)
street: Mapped[str] = mapped_column(String(255), default="", server_default="")
postal_code: Mapped[str] = mapped_column(String(30), default="", server_default="")
city: Mapped[str] = mapped_column(String(120), default="", server_default="")
state: Mapped[str] = mapped_column(String(120), default="", server_default="")
country: Mapped[str] = mapped_column(String(120), default="Deutschland", server_default="Deutschland")
is_primary: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
)
customer: Mapped[Customer] = relationship(back_populates="addresses")
class CustomerContact(Base):
__tablename__ = "customer_contacts"
id: Mapped[int] = mapped_column(primary_key=True)
customer_id: Mapped[int] = mapped_column(
ForeignKey("customers.id", ondelete="CASCADE"),
index=True,
)
first_name: Mapped[str] = mapped_column(String(100), default="", server_default="")
last_name: Mapped[str] = mapped_column(String(100), default="", server_default="")
position: Mapped[str] = mapped_column(String(120), default="", server_default="")
email: Mapped[str] = mapped_column(String(255), default="", server_default="")
phone: Mapped[str] = mapped_column(String(80), default="", server_default="")
mobile: Mapped[str] = mapped_column(String(80), default="", server_default="")
is_primary: Mapped[bool] = mapped_column(Boolean, default=False, server_default="false")
notes: Mapped[str] = mapped_column(Text, default="", server_default="")
created_at: Mapped[datetime] = mapped_column(DateTime(timezone=True), server_default=func.now())
updated_at: Mapped[datetime] = mapped_column(
DateTime(timezone=True),
server_default=func.now(),
onupdate=func.now(),
)
customer: Mapped[Customer] = relationship(back_populates="contacts")

269
backend/hermes/app/repositories/customer_repository.py Normal file
View file

@ -0,0 +1,269 @@
from datetime import UTC, datetime, timedelta
from sqlalchemy import func, select
from sqlalchemy.orm import Session, selectinload
from app.models.customer import Customer, CustomerAddress, CustomerContact
from app.schemas.customer import (
CustomerAddressCreate,
CustomerAddressUpdate,
CustomerContactCreate,
CustomerContactUpdate,
CustomerCreate,
CustomerUpdate,
)
def _email(value: object) -> str:
return "" if value is None else str(value)
def _url(value: object) -> str:
return "" if value is None else str(value)
class CustomerRepository:
@staticmethod
def get_all(db: Session) -> list[Customer]:
return list(
db.scalars(
select(Customer)
.options(
selectinload(Customer.addresses),
selectinload(Customer.contacts),
)
.order_by(Customer.created_at.desc())
)
)
@staticmethod
def get_by_id(db: Session, customer_id: int) -> Customer | None:
return db.scalar(
select(Customer)
.where(Customer.id == customer_id)
.options(
selectinload(Customer.addresses),
selectinload(Customer.contacts),
)
)
@staticmethod
def get_by_number(db: Session, customer_number: str) -> Customer | None:
return db.scalar(select(Customer).where(Customer.customer_number == customer_number))
@staticmethod
def find_number_conflict(
db: Session,
*,
customer_number: str,
exclude_customer_id: int | None = None,
) -> Customer | None:
query = select(Customer).where(Customer.customer_number == customer_number)
if exclude_customer_id is not None:
query = query.where(Customer.id != exclude_customer_id)
return db.scalar(query)
@staticmethod
def count_all(db: Session) -> int:
return db.scalar(select(func.count(Customer.id))) or 0
@staticmethod
def count_active(db: Session) -> int:
return db.scalar(select(func.count(Customer.id)).where(Customer.status == "active")) or 0
@staticmethod
def count_recent(db: Session, days: int = 30) -> int:
since = datetime.now(UTC) - timedelta(days=days)
return db.scalar(
select(func.count(Customer.id))
.where(Customer.created_at >= since)
) or 0
@staticmethod
def get_latest(db: Session, limit: int = 5) -> list[Customer]:
return list(
db.scalars(
select(Customer)
.options(selectinload(Customer.addresses), selectinload(Customer.contacts))
.order_by(Customer.created_at.desc())
.limit(limit)
)
)
@staticmethod
def create(db: Session, customer: CustomerCreate) -> Customer:
db_customer = Customer(
customer_number=customer.customer_number,
company_name=customer.company_name,
legal_name=customer.legal_name,
customer_type=customer.customer_type,
status=customer.status,
industry=customer.industry,
website=_url(customer.website),
email=_email(customer.email),
phone=customer.phone,
tax_number=customer.tax_number,
vat_id=customer.vat_id,
notes=customer.notes,
)
db_customer.addresses = [
CustomerRepository._address_from_payload(address)
for address in customer.addresses
]
db_customer.contacts = [
CustomerRepository._contact_from_payload(contact)
for contact in customer.contacts
]
CustomerRepository._normalize_primary(db_customer.addresses)
CustomerRepository._normalize_primary(db_customer.contacts)
db.add(db_customer)
db.commit()
db.refresh(db_customer)
return CustomerRepository.get_by_id(db, db_customer.id) or db_customer
@staticmethod
def update(db: Session, db_customer: Customer, customer: CustomerUpdate) -> Customer:
db_customer.customer_number = customer.customer_number
db_customer.company_name = customer.company_name
db_customer.legal_name = customer.legal_name
db_customer.customer_type = customer.customer_type
db_customer.status = customer.status
db_customer.industry = customer.industry
db_customer.website = _url(customer.website)
db_customer.email = _email(customer.email)
db_customer.phone = customer.phone
db_customer.tax_number = customer.tax_number
db_customer.vat_id = customer.vat_id
db_customer.notes = customer.notes
db_customer.addresses = [
CustomerRepository._address_from_payload(address)
for address in customer.addresses
]
CustomerRepository._normalize_primary(db_customer.addresses)
db.commit()
db.refresh(db_customer)
return CustomerRepository.get_by_id(db, db_customer.id) or db_customer
@staticmethod
def delete(db: Session, db_customer: Customer) -> None:
db.delete(db_customer)
db.commit()
@staticmethod
def get_contacts(db: Session, customer_id: int) -> list[CustomerContact]:
return list(
db.scalars(
select(CustomerContact)
.where(CustomerContact.customer_id == customer_id)
.order_by(CustomerContact.is_primary.desc(), CustomerContact.last_name)
)
)
@staticmethod
def get_contact_by_id(
db: Session,
customer_id: int,
contact_id: int,
) -> CustomerContact | None:
return db.scalar(
select(CustomerContact)
.where(CustomerContact.customer_id == customer_id)
.where(CustomerContact.id == contact_id)
)
@staticmethod
def create_contact(
db: Session,
customer_id: int,
contact: CustomerContactCreate,
) -> CustomerContact:
db_contact = CustomerRepository._contact_from_payload(contact)
db_contact.customer_id = customer_id
db.add(db_contact)
db.flush()
if db_contact.is_primary:
CustomerRepository._clear_other_primary_contacts(db, customer_id, db_contact.id)
db.commit()
db.refresh(db_contact)
return db_contact
@staticmethod
def update_contact(
db: Session,
db_contact: CustomerContact,
contact: CustomerContactUpdate,
) -> CustomerContact:
db_contact.first_name = contact.first_name
db_contact.last_name = contact.last_name
db_contact.position = contact.position
db_contact.email = _email(contact.email)
db_contact.phone = contact.phone
db_contact.mobile = contact.mobile
db_contact.is_primary = contact.is_primary
db_contact.notes = contact.notes
if db_contact.is_primary:
CustomerRepository._clear_other_primary_contacts(db, db_contact.customer_id, db_contact.id)
db.commit()
db.refresh(db_contact)
return db_contact
@staticmethod
def delete_contact(db: Session, db_contact: CustomerContact) -> None:
db.delete(db_contact)
db.commit()
@staticmethod
def _address_from_payload(address: CustomerAddressCreate | CustomerAddressUpdate) -> CustomerAddress:
return CustomerAddress(
type=address.type,
street=address.street,
postal_code=address.postal_code,
city=address.city,
state=address.state,
country=address.country,
is_primary=address.is_primary,
)
@staticmethod
def _contact_from_payload(contact: CustomerContactCreate | CustomerContactUpdate) -> CustomerContact:
return CustomerContact(
first_name=contact.first_name,
last_name=contact.last_name,
position=contact.position,
email=_email(contact.email),
phone=contact.phone,
mobile=contact.mobile,
is_primary=contact.is_primary,
notes=contact.notes,
)
@staticmethod
def _normalize_primary(items: list[CustomerAddress] | list[CustomerContact]) -> None:
primary_seen = False
for item in items:
if item.is_primary and not primary_seen:
primary_seen = True
elif item.is_primary:
item.is_primary = False
if items and not primary_seen:
items[0].is_primary = True
@staticmethod
def _clear_other_primary_contacts(db: Session, customer_id: int, contact_id: int) -> None:
contacts = db.scalars(
select(CustomerContact)
.where(CustomerContact.customer_id == customer_id)
.where(CustomerContact.id != contact_id)
)
for contact in contacts:
contact.is_primary = False

131
backend/hermes/app/schemas/customer.py Normal file
View file

@ -0,0 +1,131 @@
from datetime import datetime
from typing import Literal
from pydantic import BaseModel, ConfigDict, EmailStr, Field, HttpUrl, field_validator
CustomerStatus = Literal["lead", "active", "inactive", "blocked", "archived"]
CustomerType = Literal["company", "private", "public_sector", "partner", "supplier"]
AddressType = Literal["billing", "shipping", "primary", "other"]
def normalize_optional_text(value: object) -> str:
if value is None:
return ""
return str(value).strip()
class CustomerAddressBase(BaseModel):
type: AddressType = "primary"
street: str = Field(default="", max_length=255)
postal_code: str = Field(default="", max_length=30)
city: str = Field(default="", max_length=120)
state: str = Field(default="", max_length=120)
country: str = Field(default="Deutschland", max_length=120)
is_primary: bool = False
@field_validator("street", "postal_code", "city", "state", "country", mode="before")
@classmethod
def normalize_text(cls, value: object) -> str:
return normalize_optional_text(value)
class CustomerAddressCreate(CustomerAddressBase):
pass
class CustomerAddressUpdate(CustomerAddressBase):
pass
class CustomerAddressResponse(CustomerAddressBase):
id: int
customer_id: int
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class CustomerContactBase(BaseModel):
first_name: str = Field(default="", max_length=100)
last_name: str = Field(default="", max_length=100)
position: str = Field(default="", max_length=120)
email: EmailStr | None = None
phone: str = Field(default="", max_length=80)
mobile: str = Field(default="", max_length=80)
is_primary: bool = False
notes: str = ""
@field_validator("first_name", "last_name", "position", "phone", "mobile", "notes", mode="before")
@classmethod
def normalize_text(cls, value: object) -> str:
return normalize_optional_text(value)
class CustomerContactCreate(CustomerContactBase):
pass
class CustomerContactUpdate(CustomerContactBase):
pass
class CustomerContactResponse(CustomerContactBase):
id: int
customer_id: int
email: str
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)
class CustomerBase(BaseModel):
customer_number: str = Field(min_length=1, max_length=50)
company_name: str = Field(min_length=1, max_length=255)
legal_name: str = Field(default="", max_length=255)
customer_type: CustomerType
status: CustomerStatus
industry: str = Field(default="", max_length=120)
website: HttpUrl | None = None
email: EmailStr | None = None
phone: str = Field(default="", max_length=80)
tax_number: str = Field(default="", max_length=120)
vat_id: str = Field(default="", max_length=120)
notes: str = ""
@field_validator(
"customer_number",
"company_name",
"legal_name",
"industry",
"phone",
"tax_number",
"vat_id",
"notes",
mode="before",
)
@classmethod
def normalize_text(cls, value: object) -> str:
return normalize_optional_text(value)
class CustomerCreate(CustomerBase):
addresses: list[CustomerAddressCreate] = Field(default_factory=list)
contacts: list[CustomerContactCreate] = Field(default_factory=list)
class CustomerUpdate(CustomerBase):
addresses: list[CustomerAddressUpdate] = Field(default_factory=list)
class CustomerResponse(CustomerBase):
id: int
website: str
email: str
addresses: list[CustomerAddressResponse]
contacts: list[CustomerContactResponse]
created_at: datetime
updated_at: datetime
model_config = ConfigDict(from_attributes=True)

24
backend/hermes/app/schemas/dashboard.py Normal file
View file

@ -0,0 +1,24 @@
from pydantic import BaseModel
from app.schemas.customer import CustomerResponse
class MetricCard(BaseModel):
label: str
value: int
class EmptyWidget(BaseModel):
title: str
message: str
class DashboardSummary(BaseModel):
customers: list[MetricCard] = []
latest_customers: list[CustomerResponse] = []
users: list[MetricCard] = []
roles: list[MetricCard] = []
activities: EmptyWidget
tasks: EmptyWidget
tickets: EmptyWidget
projects: EmptyWidget