Musterlösung mit Erklärung

SCHEMA = '''
CREATE TABLE IF NOT EXISTS bestellungen (
    bestell_nr TEXT PRIMARY KEY,
    kunde TEXT NOT NULL,
    betrag NUMERIC(10, 2) NOT NULL,
    datum DATE NOT NULL
)
'''

import csv
import logging
import os
from datetime import datetime
from decimal import Decimal, InvalidOperation
from pathlib import Path

import psycopg

from schema import SCHEMA

log = logging.getLogger(__name__)


def parse_zeile(z):
    """Liefert (bestell_nr, kunde, betrag, datum) oder löst ValueError aus."""
    try:
        bestell_nr = z['bestell_nr'].strip()
        kunde = z['kunde'].strip()
        betrag = Decimal(z['betrag'].strip().replace(',', '.'))
        datum = datetime.strptime(z['datum'].strip(), '%d.%m.%Y').date()
    except (KeyError, AttributeError, InvalidOperation) as e:
        raise ValueError(f'Zeile unlesbar: {e!r}') from e
    if not bestell_nr or not kunde:
        raise ValueError('bestell_nr und kunde dürfen nicht leer sein')
    if betrag < 0:
        raise ValueError('Betrag darf nicht negativ sein')
    return bestell_nr, kunde, betrag, datum


def importiere_datei(conn, pfad):
    gut, abgelehnt = [], []
    with open(pfad, newline='', encoding='utf-8-sig') as f:
        for nr, zeile in enumerate(csv.DictReader(f, delimiter=';'), start=2):   # Zeile 1 ist die Kopfzeile
            try:
                gut.append(parse_zeile(zeile))
            except ValueError as e:
                abgelehnt.append((nr, str(e)))

    with conn.cursor() as cur:
        cur.executemany('''
            INSERT INTO bestellungen (bestell_nr, kunde, betrag, datum)
            VALUES (%s, %s, %s, %s)
            ON CONFLICT (bestell_nr) DO UPDATE
               SET kunde = EXCLUDED.kunde, betrag = EXCLUDED.betrag, datum = EXCLUDED.datum
        ''', gut)
    return len(gut), abgelehnt


def lauf(dateien):
    """Importiert eine Liste lokaler CSV-Dateien."""
    with psycopg.connect(os.environ['DATABASE_URL']) as conn:
        conn.execute(SCHEMA)
        conn.commit()
        for pfad in dateien:
            try:
                anzahl, abgelehnt = importiere_datei(conn, pfad)
                conn.commit()
            except Exception:      # bewusst breit: eine kaputte Datei darf die anderen nicht stoppen
                conn.rollback()
                log.exception('Import von %s fehlgeschlagen', Path(pfad).name)
                continue
            log.info('%s: %d Zeilen importiert, %d abgelehnt', Path(pfad).name, anzahl, len(abgelehnt))
            for nr, grund in abgelehnt:
                log.warning('%s Zeile %d: %s', Path(pfad).name, nr, grund)

from datetime import date
from decimal import Decimal

import pytest

from importer import parse_zeile


def test_gueltige_zeile():
    z = {'bestell_nr': ' A1 ', 'kunde': 'Anna', 'betrag': '12,50', 'datum': '20.10.2026'}
    assert parse_zeile(z) == ('A1', 'Anna', Decimal('12.50'), date(2026, 10, 20))


def test_ungueltiger_betrag():
    z = {'bestell_nr': 'A1', 'kunde': 'Anna', 'betrag': 'abc', 'datum': '20.10.2026'}
    with pytest.raises(ValueError):
        parse_zeile(z)

from io import BytesIO

from fastapi.responses import StreamingResponse
from openpyxl import Workbook
from openpyxl.styles import Font


def lade_bestellungen(q):
    with pool.connection() as conn:
        return conn.execute(
            'SELECT bestell_nr, kunde, betrag, datum FROM bestellungen '
            'WHERE kunde ILIKE %s ORDER BY datum DESC, bestell_nr',
            (f'%{q}%',),
        ).fetchall()


@app.get('/export.xlsx')
def excel_export(q: str = ''):
    wb = Workbook()
    ws = wb.active
    ws.title = 'Bestellungen'
    ws.append(['Bestellnr.', 'Kunde', 'Betrag', 'Datum'])
    for z in lade_bestellungen(q):
        ws.append([z['bestell_nr'], z['kunde'], float(z['betrag']), z['datum']])

    for zelle in ws[1]:
        zelle.font = Font(bold=True)
    for zeile in ws.iter_rows(min_row=2, min_col=3, max_col=3):
        zeile[0].number_format = '#,##0.00'
    for zeile in ws.iter_rows(min_row=2, min_col=4, max_col=4):
        zeile[0].number_format = 'DD.MM.YYYY'
    ws.freeze_panes = 'A2'

    puffer = BytesIO()
    wb.save(puffer)
    puffer.seek(0)
    return StreamingResponse(
        puffer,
        media_type='application/vnd.openxmlformats-officedocument.spreadsheetml.sheet',
        headers={'Content-Disposition': 'attachment; filename="bestellungen.xlsx"'},
    )


@app.get('/api/umsatz-pro-monat')
def umsatz_pro_monat():
    with pool.connection() as conn:
        return conn.execute('''
            SELECT to_char(datum, 'YYYY-MM') AS monat, SUM(betrag) AS umsatz
            FROM bestellungen
            GROUP BY monat
            ORDER BY monat
        ''').fetchall()