Kategoria: Python

Pandas i Excel – analiza raportu tankowań

Aktualizacja wcześniejszego mojego projektu i dodanie zmian polegających na uwzględnieniu wszystkich rodzajów paliw zatankowanych w danym okresie.

Program będzie wyświetlał na początku ilości zatankowanego paliwa z podziałem na jego rodzaje rodzaje np.

*** Zestawienie paliwa za okres od 1 do 31.10.2023 ***

1 GAZ LPG. 173.09 ltr

2 OLEJ NAPĘDOWY CN27102011. 6318.06 ltr

Wpisz numer opcji lub q aby wyjść:

Po wpisaniu wybranej opcji pojawia się zestawienie wybranego paliwa z podziałem na samochody np.

*** Zestawienie dla paliwa: OLEJ NAPĘDOWY CN27102011. ***

   Num. rej.    Ilość
1   AAA00123   722.00
2   AAA04225   192.02
3    AAA2X22    59.00
4    AAA52YT   578.03
5    AAA5P19    71.00
6    AAA5W41   841.00
7    AAA85K3  1209.01
8    AAA85YY   879.00
9    AAAFX95   826.00
10   AAAPM20    56.00
11   AAARG00   187.00
12   AAARU44   698.00

Zakupione paliwo:  6318.06 ltr

Wpisz numer rej. lub p aby powrócić do głównego menu:

Po wybraniu dowolnego fragmentu numeru rejestracyjnego, bez rozróżnienia wielkości liter pokazuje się kolejny ekran, np.

*** Tankowania dla pojazdu: AAARG00 ***

Data         Ilość        Netto        Brutto
02-11-2023   43.00 ltr    5.12           6.30
06-11-2023   42.00 ltr    5.24           6.44
07-11-2023   20.00 ltr    5.29           6.51
10-11-2023   27.00 ltr    5.20           6.40
14-11-2023   27.00 ltr    5.17           6.36
21-11-2023   28.00 ltr    5.09           6.26

 Suma tankowań dla pojazdu: 187.00 ltr

Wpisz p aby powrócić do wcześniejszego menu:

Pobrany plik programu Excel zawiera między innymi następujące kolumny: „Dane kontrahenta”, „Numer WZ”, „Data”, „Godzina”, „Licznik”, „Numer rejestracyjny”, „Numer karty”, „Nazwa towaru”, „Odbiorca” itd.

Każdy wiersz zawiera dane o konkretnym tankowaniu.

Korzystać będę tylko z danych z paru kolumn tzn. „Data”, „Numer rejestracyjny”, „Nazwa towaru” – zawierająca rodzaj pobranego paliwa, „Ilość” – ilość zatankowanego paliwa, oraz „Cena brutto” i „Cena netto”.

W tym celu tworzę klasę Reports(), która będzie zawierać wczytane raporty tankowań:

#!/usr/bin/env python3
# -*- coding: utf-8 -*-
'''
@ author: Sławomir Kwiatkowski
@ date: 2023.11.18
'''
class Reports():
   pass

if __name__ == '__main__':
    Reports()

W metodzie __init__() dokonuję inicjalizacji dla modułu colorama, następnie metoda: load_last_report() wczytuje ostatni plik Excela zawierający najnowsze raporty tankowań, a następnie uruchomiona jest metoda main() klasy tzn.

def __init__(self):
    colorama.init(autoreset=True)
    last_report = self.load_last_report()
    self.main(last_report)

Metoda load_last_report() pobiera najnowszy plik excela z podkatalogu 'Dane’ i zwraca ramkę Pandas tzn.

def load_last_report(self):
    current_dir = os.path.dirname(os.path.abspath(__file__))
    data_dir = os.path.join(current_dir, 'Dane')
    reports = glob.glob(os.path.join(data_dir, '*.xlsx'))
    last_report = max(reports, key=os.path.getctime)
    path_to_last_report = os.path.join(data_dir, last_report)
    return  pd.read_excel(path_to_last_report)

Należy uzupełnić wymagane importy tzn.

import os
import glob
import colorama
import pandas as pd

Wydzieliłem prywatną metodę do „czyszczenia” konsoli tzn.

def _clear_screen(self):
    os.system('cls' if os.name == 'nt' else 'clear')

W metodzie main() klasy Reports wyświetlam krótki nagłówek copyright, a następnie wypisuję zsumowane paliwo, w zależności od rodzaju paliwa tzn.

    def main(self, report):
        self._clear_screen()
        copyright = {'author:': 'Sławomir Kwiatkowski', 'Date:': '18-11-2023' }
        for item in copyright:
            print(f'{colorama.Fore.RED}  {item} {copyright[item]}'.rjust(120)) 
        self.main_header(report['Data'].min(), report['Data'].max())
        fuel_grp = report.groupby('Nazwa towaru')
        fuel_grp_sums = fuel_grp['Ilość'].sum()
        for i, sum in enumerate(fuel_grp_sums.items(), start=1):
            print(i, sum[0], f'{sum[1]:.2f}',  'ltr', '\n')

W metodzie main() następnie wczytuję opcję wybraną przez użytkownika tzn. numer wybranego paliwa, dla którego wyświetlić szczegółowe dane lub „q” aby wyjść z programu tzn.

    while True:
        choice = input('Wpisz numer opcji lub q aby wyjść: ')
        if choice == 'q': 
            self._clear_screen()
            raise SystemExit
        if choice.isnumeric() and choice!='0':
            try:
                fuel= fuel_grp_sums.index[int(choice)-1]
                self.fuel_description(report, fuel)
                break
            except:
                pass
        self.main(report)

Metoda main() zawiera też wywołanie nagłówka tzn. main_header(), który wyświetla napis o zakresie dat, z którego pochodzą dane tankowań tzn. data pierwszego tankowania i data ostatniego tankowania:

def main_header(self, start_date, end_date):
    print(colorama.Fore.YELLOW +'*** Zestawienie paliwa za okres od {} do {}.{}.{} ***\n'
          .format(start_date.day,
                  end_date.day,
                  end_date.month,
                  end_date.year
                  ))

Jeśli użytkownik wybierze numer paliwa, dla którego chce wyświetlić szczegółowy opis, to wyświetla się kolejny ekran, zawierające dane o sumach tankowań wybranego paliwa dla poszczególnych pojazdów tzn.

def fuel_description(self, report, fuel):
    self._clear_screen()
    print(colorama.Fore.YELLOW + f'*** Zestawienie dla paliwa: {fuel} ***\n')
    filt = report['Nazwa towaru'] == fuel
    cars_group = report[filt].groupby('Numer rejestracyjny')
    cars_group_sums = cars_group['Ilość'].sum().reset_index()
    cars_group_sums.index +=1
    cars_group_sums.rename(columns={'Numer rejestracyjny': 'Num. rej.'}, inplace=True)
    print(cars_group_sums, '\n')
    print('Zakupione paliwo: ', cars_group_sums.sum()[1], 'ltr \n')

Metoda ta zawiera również możliwość wybrania fragmentu numeru rejestracyjnego, aby wyświetlić szczegóły tankowań danego paliwa dla wybranego auta lub przejście do wcześniejszego ekranu po wciśnięciu „p”.

    while True:
        choice = input('Wpisz numer rej. lub p aby powrócić do głównego menu: ')
        if choice == 'p': 
            break
        for key, value in cars_group['Ilość']:
            if choice.upper() in key:
                self.car_description(key, value, report, fuel)
                break
        self.fuel_description(report, fuel)    
    self.main()

Po wybraniu dowolnego fragmentu numeru rejestracyjnego, metoda car_description() wyświetla szczegółowe dane dla danego auta tzn. datę tankowania, ilość zatankowanego paliwa oraz cenę netto i brutto tzn.

def car_description(self, key, value, report, fuel):
    self._clear_screen()
    print(colorama.Fore.YELLOW + f'*** Tankowania dla pojazdu: {key} ***\n')
    header = ['Data', 'Ilość', 'Netto', 'Brutto']
    for item in header:
        print(f'{item:<13}', end="")
    print()
    total = 0
    line = 1
    for i, v in zip(value.index, value.values):
        output = f'{report.loc[i, "Data"]:%d-%m-%Y}' \
                    f'{v:>8.2f} ltr' \
                    f'{report.loc[i, "Cena netto"]:>8.2f}' \
                    f'{report.loc[i, "Cena brutto"]:>15.2f}'
        total += v
        print(colorama.Fore.BLACK + colorama.Back.WHITE + output
              ) if line % 2 else print(output)
        line += 1
    print(f'\n Suma tankowań dla pojazdu: {total:.2f} ltr \n')
            
    while True:
        choice = input('Wpisz p aby powrócić do wcześniejszego menu: ')
        if choice == 'p': 
            break
        self.car_description(key, value, report, fuel)
    self.fuel_description(report, fuel)

Własne uprawnienia w DRF

Aby utworzyć własne uprawnienia dla niestandardowego użytkownika utworzonego wcześniej tworzę klasę dziedziczącą po klasie BasePermission.

Utworzę dwie klasy uprawnień:

  • Klasę umożliwiającą dostęp użytkownikom, którzy są klientami (atrybut role ma wartość client). Dostawcy nie mają uprawnień do zasobu.
  • Klasę umożliwiającą użytkownikom, którzy są dostawcami tylko możliwość odczytu.

Na wstępie, w pliku permissions.py aplikacji contracts importuję klasę BasePermission, oraz listę „bezpiecznych metod” tzn. umożliwiających wylistowanie zasobu, bez możliwości edycji, usuwania czy dodania nowego kontraktu.

from rest_framework.permissions import BasePermission, SAFE_METHODS


"""Custom permissions classes"""

Klasa IsClient() umożliwia dostęp tylko klientom. Pozostali użytkownicy zobaczą komunikat message informujący o braku dostępu do zasobu:

class IsClient(BasePermission):
    message = "Only clients can access"

    def has_permission(self, request, view):
        if request.user.role == "client":
            return True
        return False

Klasa IsClientOrReadOnly() umożliwia użytkownikom, którzy są klientami możliwość dodania, kasowania i edycji, pozostali użytkownicy maja możliwość wyświetlenia danych. tzn.:

class IsClientOrReadOnly(BasePermission):
    message = "Only clients can modify, contractors can read only"

    def has_permission(self, request, view):
        if request.method in SAFE_METHODS:
            return True
        return request.user.role == "client"

Utworzone klasy uprawnień umieszczam w widokach np.

class ContractViewSet(viewsets.ModelViewSet):
    permission_classes = [IsAuthenticated, IsClientOrReadOnly]
    serializer_class = ContractSerializer

Django Rest Framework – testowanie API

W tym wpisie opiszę testowanie API dla utworzonego kastomowego użytkownika. (wcześniejsze wpisy – >>część 1<< oraz >>część 2<<)

Dostęp do widoku ContactUserCreateRetrieveViewSet umożliwiającego tworzenie nowego użytkownika oraz wyświetlanie konkretnego użytkownika jest zarejestrowany w pliku urls.py aplikacji users tzn.

from rest_framework.routers import DefaultRouter
from .views import ContractUserCreateRetrieveViewSet

router = DefaultRouter()

router.register("", ContractUserCreateRetrieveViewSet, basename="user")

urlpatterns = router.urls

Jako parametr basename zdefiniowałem jako user, do której to nazwy będę się odwoływał podczas testów.

Plik urls.py aplikacji users jest następnie importowany w pliku urls.py projektu tzn.

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api-auth/", include("rest_framework.urls")),
    path("api/user/", include("users.urls")),
]

Kompletny punkt dostępowy do tworzenia użytkowników będzie więc miał postać /api/user/

Tworzenie użytkownika i wyświetlanie konkretnego użytkownika będzie realizowane przez konkretne metody z pliku views.py aplikacji users tzn. tworzenie – funkcja create(), wyświetlanie użytkownika – funkcja retrieve()

from django.shortcuts import get_object_or_404
from rest_framework import viewsets, mixins
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework import status
from .models import ContractUser
from .serializers import ContractUserSerializer

class ContractUserCreateRetrieveViewSet(
    mixins.CreateModelMixin,
    mixins.RetrieveModelMixin,
    viewsets.GenericViewSet,
):
    serializer_class = ContractUserSerializer
    queryset = ContractUser.objects.all()

    def retrieve(self, request, pk):
        permission_classes = [IsAuthenticated]
        queryset = ContractUser.objects.filter(id=request.user.id)
        user = get_object_or_404(queryset, pk=pk)
        serializer = ContractUserSerializer(user)
        return Response(serializer.data)

    def create(self, request):
        serializer = ContractUserSerializer(data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        else:
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

Klasa ContractUserCreateRetrieveViewSet dziedziczy po podstawowej klasie GenericViewSet oraz po klasach mixins, które dodają wymagane funkcjonalności – tworzenie i wyświetlanie użytkownika.

Funkcja create() serializuje podane dane i po sprawdzeniu, czy są one poprawne zapisuje je w bazie i zwraca obiekt Response i odpowiedni kod statusu.

Funkcja retrieve() sprawdza czy użytkownik jest zalogowany i czy filtruje użytkowników, aby zwracany queryset zawierał tylko tego użytkownika, który jest obecnie zalogowany, a następnie zwraca obiekt Response i odpowiedni kod statusu.

W pliku signals.py aplikacji users modyfikuję pole email udostępnione przez klasę AbstractUser, aby pole to było wymagane podczas tworzenia modelu ContractUser tzn.

from .models import ContractUser
from django.dispatch import receiver


@receiver(pre_save, sender=ContractUser)
def create_inactive_user(sender, instance, **kwargs):
    instance.is_active = False
    instance._meta.get_field("email").blank = False
    instance._meta.get_field("email").null = False

Jeśli chcielibyśmy, aby pole email było unikalne, to możemy ustawić parametr unique jako True, tzn.

    instance._meta.get_field("email")._unique = True

Wówczas jeśli użytkownik wpisze email, który już jest w bazie to wystąpi błąd walidacji. Niestety dzięki temu będzie wiadomo, że użytkownik o takim adresie email jest już obecny w bazie.

Aby przetestować tworzenie i wyświetlanie użytkownika tworzę katalog tests w aplikacji users.

Do niego przenoszę plik tests.py utworzony w poprzednim artykule, który zawierał testy modelu i zmieniam mu nazwę na test_model.py

W utworzonym katalogu tests tworzę nowy plik o nazwie test_api.py, który będzie zawierał testy api dla aplikacji users tzn.

from rest_framework.test import APITestCase
from rest_framework import status
from django.urls import reverse
from ..models import ContractUser


class ContractUserApiTestCase(APITestCase):
    """Testing ContractUser API"""

    def test_create_user(self):
        endpoint = reverse("user-list")
        user_data = {
            "username": "user_1",
            "password": "123456789",
            "email": "user_1@company.com",
        }
        response = self.client.post(endpoint, data=user_data, format="json")
        self.assertEqual(response.status_code, status.HTTP_201_CREATED)

    def test_retrieve_user(self):
        user = ContractUser.objects.create(
            username="user1", password="12345", email="user1@company.com"
        )
        self.client.force_authenticate(user)
        endpoint = reverse("user-detail", args=[user.id])
        response = self.client.get(endpoint, format="json")
        self.assertEqual(response.status_code, status.HTTP_200_OK)

W metodzie test_create_user() tworzę endpoint przy użyciu funkcji reverse(). Funkcja reverse() bierze jako parametr basename nadany w pliku urls.py wraz z sufiksem -list. Następnie tworzę dane dla testowego użytkownika i przy wykorzystaniu domyślnego klienta obecnego w klasie APITestCase dokonuję zapytania przy użyciu metody POST. Asercja sprawdza, czy kod statusu odebranej odpowiedzi to 201, gdy pomyślnie utworzono użytkownika, czy też wynosi on 400 w przeciwnym przypadku.

W metodzie test_retrieve_user() tworzę testowego użytkownika i następnie zalogowuję go za pomocą funkcji force_authenticate(). Funkcja reverse() jako parametr bierze wartość basename nadany w pliku urls.py wraz z sufiksem -detail. Jako dodatkowy parametr załączam id zalogowanego użytkownika. Asercja sprawdza, czy kod statusu odpowiedzi to 200, gdy zapytanie zakończyło się sukcesem czy też status wynosi 400 w przeciwnym przypadku.

Django – testowanie modelu

Aby przetestować kastomowy model ContractUser utworzony w poprzednim artykule, w pliku tests.py aplikacji users importuję klasę TestCase z modułu django.test oraz testowany model tzn.

Tworzę nową klasę ContractUserTestCase, która dziedziczy po klasie TestCase.

W metodzie setUp() tworzę dwóch tymczasowych użytkowników user1 i user2.

W metodzie test_create_user() przeprowadzam 1 test składający się z kilku asercji, w szczególności sugerowany przez narzędzie coverage test tekstowej reprezentacji utworzonej instancji tzn.

class ContractUserTestCase(TestCase):
    def setUp(self):
        self.user1 = ContractUser.objects.create(
            username="user_1",
            password="12345",
            role="contractor"
        )
        self.user2 = ContractUser.objects.create(
            username="user_2",
            password="12345")

    def test_create_user(self):
        self.assertEqual(self.user1.username, "user_1")
        self.assertEqual(self.user2.username, "user_2")
        self.assertEqual(self.user1.role, "contractor")
        self.assertEqual(self.user2.role, "client")
        self.assertEqual(str(self.user1), "user_1")
        self.assertEqual(str(self.user2), "user_2")

W podobny sposób testuję pozostałe modele z aplikacji contracts.

W modelu Contract obecny jest także kastomowy walidator, który sprawdza, czy data dostawy nie jest wcześniejsza niż data złożenia zlecenia.

część modelu Contract z models.py aplikacji contracts:

class Contract(models.Model):
    class Meta:
        ordering = ("-date_of_order",)

    class StatusChoices(models.TextChoices):
        OPEN = "open"
        ACCEPTED = "accepted"
        CANCELED = "cancelled"

    class DeliveryDateValidator:
        def validate(value):
            if value < timezone.now().date():
                raise validators.ValidationError(
                    "Date of delivery can't be earlier than date of order"
                )
            else:
                return value

    date_of_order = models.DateField(default=timezone.now)

    date_of_delivery = models.DateField(
        validators=(DeliveryDateValidator.validate,)
)

             ..........

Aby przetestować kastomowy walidator tworzę osobny test, który sprawdza, czy zostanie wywołany wywołany wyjątek ValidationError oraz czy zostanie zwrócona wpisana data, gdy podan0 poprawną datę (nie starszą od daty zamówienia).

    def test_delivery_date_validator(self):
        with self.assertRaises(ValidationError):
            self.contract.full_clean()
        self.contract.date_of_delivery = datetime.strptime(
            "2200-01-01", "%Y-%m-%d"
        ).date()
        self.assertEqual(
            self.contract.DeliveryDateValidator.validate(
                self.contract.date_of_delivery
            ),
            self.contract.date_of_delivery,
        )

Django rest framework – tworzenie użytkowników

Na wstępie tworzę nową aplikację o nazwie users tzn.

python manage.py startapp users

W pliku models.py tworzę niestandardowego użytkownika, który dziedziczy po klasie AbstractUser. Dodaję pole wyboru role, dzięki czemu użytkownik będzie należał albo do zleceniodawców, albo do zleceniobiorców (client lub contractor). 

from django.contrib.auth.models import AbstractUser
from django.db import models


class ContractUser(AbstractUser):
    class RoleChoices(models.TextChoices):
        CLIENT = "client"
        CONTRACTOR = "contractor"

    role = models.CharField(
        max_length=10,
        choices=RoleChoices.choices, default=RoleChoices.CLIENT
    )

    def __str__(self):
        return self.username

Nowy model user należy zarejestrować w pliku settings.py projektu, tzn.

AUTH_USER_MODEL = "users.ContractUser"

W aplikacji users tworzę plik serializers.py zawierający serializer nowego modelu użytkownika tzn.

from django.contrib.auth.hashers import make_password
from rest_framework import serializers
from .models import ContractUser


class ContractUserSerializer(serializers.ModelSerializer):
    password = serializers.CharField(
        min_length=8,
        write_only=True,
        style={"input_type": "password"}
    )

    class Meta:
        model = ContractUser
        fields = [
             "id",
             "username",
             "password",
             "email",
             "role"
        ]

    def create(self, validated_data):
        validated_data["password"] = make_password(validated_data["password"])
        return super().create(validated_data)

Serializer zawiera pole tekstowe umożliwiające wprowadzenie pola tekstowego dla hasła. Pole to zawiera parametr write_only ustawione na wartość True, aby możliwy był tylko zapis, a nie odczyt tego pola. Pole fields klasy wewnętrznej Meta zawiera listę wszystkich dostępnych pól do serializacji. Metoda create() tworzy hasło na podstawie wprowadzonych przez użytkownika danych przy użyciu funkcji make_password().

W pliku urls.py aplikacji users definiuję punkt dostępowy tzn.

from rest_framework.routers import DefaultRouter
from .views import ContractUserCreateRetrieveViewSet

router = DefaultRouter()

router.register("", ContractUserCreateRetrieveViewSet, basename="user")

urlpatterns = router.urls

Punkt dostępowy dla aplikacji users dołączam do głównego pliku urls.py projektu:

from django.contrib import admin
from django.urls import path, include

urlpatterns = [
    path("admin/", admin.site.urls),
    path("api/user/", include("users.urls")),
]

Punkt dostępowy /api/user/ będzie umożliwiał tworzenie użytkownika i podgląd danych dla danego użytkownika. W tym celu w pliku views.py tworzę klasę ContractUserCreateRetrieveViewSet(). Aby ograniczyć liczbę dostępnych metod nie korzystam z klasy ModelViewSet, ale dziedziczę po klasie GenericViewSet oraz odpowiednich klasach Mixin.

from django.shortcuts import get_object_or_404
from rest_framework import viewsets, mixins
from rest_framework.permissions import IsAuthenticated
from rest_framework.response import Response
from rest_framework import status
from .models import ContractUser
from .serializers import ContractUserSerializer

class ContractUserCreateRetrieveViewSet(
    mixins.CreateModelMixin,
    mixins.RetrieveModelMixin,
    viewsets.GenericViewSet,
):
    serializer_class = ContractUserSerializer
    queryset = ContractUser.objects.all()

    def retrieve(self, request, pk):
        permission_classes = [IsAuthenticated]
        queryset = ContractUser.objects.filter(id=request.user.id)
        user = get_object_or_404(queryset, pk=pk)
        serializer = ContractUserSerializer(user)
        return Response(serializer.data)

    def create(self, request):
        serializer = ContractUserSerializer(data=request.data)
        if serializer.is_valid():
            serializer.save()
            return Response(serializer.data, status=status.HTTP_201_CREATED)
        else:
            return Response(serializer.errors, status=status.HTTP_400_BAD_REQUEST)

Aby nowo utworzony użytkownik miał status nieaktywny skorzystałem z sygnału pre_save. W tym celu utworzyłem nowy plik signals.py w aplikacji users tzn.

from django.db.models.signals import pre_save
from .models import ContractUser
from django.dispatch import receiver

@receiver(pre_save, sender=ContractUser)
def create_inactive_user(sender, instance, **kwargs):
    instance.is_active = False

Sygnał należy zaimportować w funkcji ready() klasy UsersConfig() z pliku apps.py

from django.apps import AppConfig


class UsersConfig(AppConfig):
    default_auto_field = "django.db.models.BigAutoField"
    name = "users"

    def ready(self):
        import users.signals

Tkinter app – pobieranie danych ze zdalnego API

Kod źródłowy dostępny na GitHub

W tym przykładzie dane pobrane do programu pochodzą z archiwum cen hurtowych Orlenu.

W związku z tym, że zmianie uległ sposób pobierania danych ze strony, musiałem zmienić także sposób działania mojego programu (poprzednia wersja wykorzystywała przeszukiwanie strony w celu odnalezienia na niej odpowiednich wartości – web scraping). Obecnie konieczne stało się skorzystanie z API, w celu uzyskania danych o cenach.

Aby ustalić, jak ma wyglądać poprawne zapytanie do API Orlenu skorzystałem z zakładki Sieć trybu developera przeglądarki.

Program będzie umożliwiał pobranie danych o cenach dla paliw, przy czym możliwe będzie wyświetlenie danych i wykresu z podziałem na konkretne lata i miesiące.

Projekt będzie składał się z głównego pliku programu: orlen-prices.py oraz pliku utils.py zawierającego funkcję do pobrania odpowiedzi z zewnętrznego API oraz funkcję do konwersji danych w formacie json do odpowiedniej ramki danych Pandas.

Pobrane dane będą w prosty sposób buforowane, aby ograniczyć liczbę zapytań do API.

Na początek opiszę funkcję fetch_data() z pliku utils.py odpowiadającą za pobranie odpowiedzi z zewnętrznego serwera. Jeśli zapytanie zakończy się sukcesem, funkcja zwraca odpowiedź w formacie json. Jeśli natomiast wystąpi wyjątek, to funkcja zwraca wartość None.

def fetch_data(product_id, current_date):
        url = f'https://tool.orlen.pl/api/wholesalefuelprices/'\
                    f'ByProduct?productId={product_id}'\
                        f'&from=2000-01-01'\
                            f'&to={current_date}'
        try:
            response = requests.get(url)
        except requests.RequestException:
            return None
        else:
            return response.json()

Następna funkcja z pliku utils.py dokonuje konwersję danych w formacie json do ramki danych Pandas. Wynikowa ramka zawierać będzie tylko dwie kolumny, pozostałe dane są pominięte. Kolumna effectiveDate będzie zawierać wartości typu datetime oraz będzie ustawiona jako index (w celu ułatwienia przeszukiwania po dacie). Kolumna value będzie miała format int. Cała ramka będzie natomiast posortowana względem wartości index.

def json_to_df(data):
    df = pd.json_normalize(data)
    df = df[['effectiveDate', 'value']]
    df['effectiveDate'] = pd.to_datetime(df['effectiveDate'])
    df['value'] = df['value'].astype(int)
    df.set_index(df['effectiveDate'], inplace=True)
    df.sort_index(inplace=True)
    return df

Główny plik zawiera klasę MainWindow odpowiedzialną za wyświetlenie okna programu i wszystkich widgetów, która jest zainicjowana standardowo:

if __name__ == "__main__":
    root = tk.Tk()
    root.protocol("WM_DELETE_WINDOW", MainWindow._exit)
    app = MainWindow(root)
    app.mainloop()

Powyższy kod tworzy instancję klasy MainWindow oraz ustala funkcję, która umożliwia zamknięcie programu – jest to funkcja statyczna klasy:

    @staticmethod
    def _exit():
        root.destroy()
        root.quit()

Klasa MainWindow będzie dziedziczyć po klasie Frame biblioteki tkinter:

import tkinter as tk
class MainWindow(tk.Frame):
    pass

Na wstępie definiuję stałe i zmienne klasy np.

  • background – definiuje kolor tła widgetu TreeView
  • price_data – słownik zawierający pobrane dane z API
  • current_date – zmienna zawierająca bieżącą datę
  • months – lista skrótów nazw miesięcy
  • products – słownik zawierający nazwy produktów i ich kody
    bacground = '#EEEEEE'
    price_data = {}
    current_date = date.today()
    months = calendar.month_abbr
    products = {
        'Eurosuper95': 41,
        'SuperPlus98': 42,
        'Bio100': 47,
        'Arktyczny2': 44,
        'Ekodiesel': 43,
        'GrzewczyEkoterm': 46,
        'MiejskiSuper' : 45
    }

W metodzie __init__() inicjuję wartości z klasy rodzica, a następnie uruchamiam metodę create_UI() odpowiedzialną za wyświetlenie widżetów.

    def __init__(self, root):
        super().__init__(root)
        self.create_UI(root)

metoda createUI() ustala rozmiar okna programu na zmaksymalizowany, ustawiany jest tytuł okna a następnie inicjalizowane są odpowiednie widgety tzn.

  • metoda show_menu() ustawia pasek menu
  • metoda show_product_combobox() ustawia pole wyboru produktu
  • metoda show year_combobox() ustawia pole wyboru roku
  • metoda show_radiobuttons() ustawia przyciski radio do wyboru miesiąca
  • metoda show_chart_button() ustawia przycisk wyświetlający wykres
  • metoda show_msg_label() ustawia etykietę pola statusu
  • metoda show_data_table() ustawia widget wyświetlający dane o cenach
    def create_UI(self, root):
        root.state('zoomed')
        root.title('Orlen wholesale prices')
        self.show_menu(root)
        self.show_product_combobox(root)
        self.show_year_combobox(root)
        self.show_radiobuttons(root)
        self.show_chart_button(root)
        self.show_msg_label(root)
        self.show_data_table(root)

Z menu będzie możliwe opcjonalne zamknięcie programu poprzez opcję File/Exit tzn.

    def show_menu(self, root):
        # Menu bar
        menu_bar = tk.Menu(root)
        root.config(menu=menu_bar)
        # File menu 
        file_menu = tk.Menu(menu_bar, tearoff=0)
        file_menu.add_command(label='Exit', command=self._exit)
        menu_bar.add_cascade(label='File', menu=file_menu)

Metoda show_product_combobox() inicjalizuje widget Combobox, który będzie odpowiedzialny za wybór produktu, dla którego chcemy prześledzić zmianę cen. Wybór będzie przekazywany za pomocą zmiennej product_sv – obiekt StringVar. Zanim użytkownik wybierze odpowiedni produkt na widżecie jest wyświetlana podpowiedź.

    def show_product_combobox(self, root):
        self.product_sv = tk.StringVar(value='')
        product_cb = ttk.Combobox(root, textvariable=self.product_sv)
        product_cb['values'] = [x for x in self.products]
        product_cb.set('Product')
        product_cb.state = 'readonly'
        product_cb.bind('<<ComboboxSelected>>', self.enable_show_chart_btn)
        product_cb.grid(row=0, column=0, padx=10, pady=10)

Gdy użytkownik wybierze odpowiedni produkt to uruchamiana jest powiązana metoda: enable_show_chart_button(), gdyż domyślnie przycisk rysowania wykresu jest nieaktywny. Początkowo sprawdzany jest warunek, czy przycisk jest nieaktywny, następnie czy zostały wybrane produkt i rok. Jeśli wszystkie te warunki zostały spełnione to stan przycisku zmienia się na aktywny.

    def enable_show_chart_btn(self, sender):
        if str(self.show_chart_btn['state']) == 'disabled' and self.product_sv.get() != 'Product' and self.year_sv.get() != 'Year':
            self.show_chart_btn['state'] = '!disabled'

Podobnie wygląda metoda wyboru roku, dla którego chcemy przeglądać dane. Wartości lat, z których można dokonywać wyboru są z zakresu od 2004 do obecnego roku, przy czym bieżący rok jest najwyżej na liście.

    def show_year_combobox(self, root):
        self.year_sv = tk.StringVar(value='')
        year_cb = ttk.Combobox(root, textvariable=self.year_sv)
        year_cb['values'] = [x for x in range(self.current_date.year, 2004-1, -1)]
        year_cb.set('Year')
        year_cb.state = 'readonly'
        year_cb.bind('<<ComboboxSelected>>', self.enable_show_chart_btn)
        year_cb.grid(row=0, column=1, padx=10, pady=10)

Pola radio służące do wyboru konkretnego miesiąca są osadzone w osobne ramce, a wybrana wartość jest przekazywana przez zmienną radio_sv. Jako że calendar.month_abbr zwraca listę, gdzie index 1 odpowiada za wartość stycznia itd, to wartość o indeksie 0 zawiara pusty łańcuch. Wykorzystałem to do umieszczenia opcji domyślnej – All – gdy chcemy wyświetlić wartości dla wszystkich miesięcy danego roku. Funkcja enumerate() użyta na liście umożliwiła umieszczenie odpowiedniego widgetu radio w kolejnej kolumnie.

    def show_radiobuttons(self, root):
        # Frame for Radiobuttons
        rb_frame = tk.Frame(root)
        rb_frame['borderwidth'] = 1
        rb_frame['relief']='groove'
        rb_frame.grid(row=0, column=2, sticky='w')
        # Radiobuttons
        self.radio_sv = tk.StringVar(value='All')
        for i, month in enumerate(self.months):
            if month=="":
                month_rb = tk.Radiobutton(
                    rb_frame, 
                    text='All',
                    value='All', 
                    variable=self.radio_sv)
            else:
                month_rb = tk.Radiobutton(
                    rb_frame, 
                    text=month,
                    value=month, 
                    variable=self.radio_sv)
            month_rb.grid(row=0, column=i)

Przycisk uruchamiający rysowanie wykresu jest domyślnie nieaktywny. Dopiero wybranie produktu i roku zmienia stan przycisku na aktywny. Przyciśnięcie przycisku uruchamia metodę show_chart_btn_clicked()

    def show_chart_button(self, root):
        self.show_chart_btn = ttk.Button(
            root,
             text='Show Chart',
             command=self.show_chart_btn_clicked)
        self.show_chart_btn['state'] = 'disabled'
        self.show_chart_btn.grid(row=0, column=3, padx=10, sticky='w')

Dane w postaci liczbowej umieszczone są w obiekcie Treeview. Dane te są pobierane dla konkretnego produktu i daty po wciśnięciu przycisku. 

    def show_data_table(self, root):
        # Show Data Frame
        data_frame = tk.Frame(root)
        data_frame['borderwidth'] = 1
        data_frame['relief']='groove'
        data_frame.grid(row=1, column=0, columnspan=2, sticky=tk.N)
        # Show Data Table
        nametofont("TkHeadingFont").configure(weight='bold')
        self.table = ttk.Treeview(data_frame, 
                    show='headings')
        scrollbar = ttk.Scrollbar(data_frame, 
                                orient='vertical',
                                command=self.table.yview)
        scrollbar.grid(row=0, column=1, sticky=tk.NS)
        
        self.table.configure(yscrollcommand=scrollbar.set)

        self.table['columns'] = ('Date', 'Price')
        self.table.column('Date', anchor=tk.CENTER)
        self.table.column('Price', anchor=tk.CENTER)

        self.table.heading('Date', text='Date', anchor=tk.CENTER)
        self.table.heading('Price', text='Price [PLN]', anchor=tk.CENTER)

        self.table.tag_configure('odd', background=self.bacground)
        self.table.configure

        self.table.grid(row=0, column=0)

Program zawiera pasek statusu, w którym umieszczane są komunikaty o poprawnym wczytaniu danych, braku danych dla wybranego produktu w wybranym okresie lub o błędach związanych np. z brakiem połączenia z internetowego.

    def show_msg_label(self, root):
        self.msg = tk.StringVar()
        msg_label = tk.Label(root, 
                                relief='groove',
                                anchor='w',
                                textvariable=self.msg)
        msg_label.grid(row=3, column=0, columnspan=6, sticky='ews')
        root.grid_rowconfigure(3, weight=1)
        root.grid_columnconfigure(5, weight=1)

Wybranie przycisku rysowania wykresu robi następujące rzeczy:

  • kasuje komunikat pola statusu
  • jeśli dane o produkcie są już wcześniej pobrane nie dokonuje zapytania do zewnętrznego API
  • jeśli dane są jeszcze nie pobrane to następuje próba pobrania ich z serwera. Gdy próba się nie powiedzie to jest wyświetlany komunikat o błędzie pobierania
  • dane zostają przefiltrowane na podstawie wybranych przez użytkownika wartości daty
  • przefiltrowane dane są parametrami przekazanymi do metod: _set_data_table(), która aktualizuje dane liczbowe oraz _show_chart(), która wyświetla wykres
def show_chart_btn_clicked(self):
        self.msg.set('')
        product_id = self.products[self.product_sv.get()]
        if product_id in self.price_data:
            data = self.price_data[product_id]
        else:
            data = fetch_data(product_id, self.current_date)
            if data is not None:
                self.price_data[product_id] = data
            else:
                self.msg.set('Error fetching data')
                return 
        product_df = json_to_df(data)
        year = self.year_sv.get()
        month = self.radio_sv.get()
        self.table.delete(*self.table.get_children())
        try:
            if month == 'All':
                chart_df = product_df.loc[year]
            else: 
                month_int = datetime.strptime(month, '%b').month
                chart_df = product_df.loc[f'{year}-{month_int}']
        except:
            self.msg.set('No data for selected date')
            for widget in self.chart_canvas.winfo_children():
                widget.destroy()
        else:
            self.msg.set('OK')
            self._set_data_table(chart_df[::-1])
            self._show_chart(root, chart_df, year, month)

Aktualizacja danych w obiekcie Treeview wyświetlającym dane liczbowe o cenach:

def _set_data_table(self, df):
        for i, row in enumerate(df.iloc):
            date, value = row
            if i % 2:
                self.table.insert(
                    parent='', index=i, text='', 
                    values=(date.date(), value),
                    tag='odd')
            else:
                self.table.insert(
                    parent='', index=i, text='', 
                    values=(date.date(), value))

Wyświetlenie wykresu:

def _show_chart(self, root, df, year, month):
        # Show Chart Frame
        self.chart_canvas = tk.Frame(root)
        self.chart_canvas['borderwidth'] = 1
        # self.chart_canvas['relief']='groove'
        self.chart_canvas.grid(row=1, column=2)
        # Chart
        date = [str(x.date()) for x in df['effectiveDate']]
        price = df['value'].tolist()
        fig, ax = plt.subplots()
        if month == 'All':
            month = ''
        ax.set_title(
            f'Orlen wholesale prices \
                {self.product_sv.get()} [PLN/m3] - {month} {year}')
        ax.set_xlabel('Date')
        ax.set_ylabel('Price [PLN]')
        fig.autofmt_xdate()
        ax.grid(True)
        ax.xaxis.set_major_locator(plt.MaxNLocator(12))
        ax.yaxis.set_major_locator(plt.MaxNLocator(12))
        textstr='(c) S.Kwiatkowski'
        props = dict(boxstyle='round', alpha=0.5)
        ax.plot(date, price, c='#CA3F62')
        ax.text(0.8, 0.95, textstr, transform=ax.transAxes, fontsize=8,
            verticalalignment='top', bbox=props)
        canvas = FigureCanvasTkAgg(fig, master=self.chart_canvas)
        canvas.draw()
        canvas.get_tk_widget().grid(row=1, column=2, columnspan=2, rowspan=2, sticky='nswe')

Bulma – modal komponent + React +Flask-RESTful

We wpisie opiszę, jak uaktywnić komponent modal z frameworka Bulma, aby po wciśnięciu przycisku kasowania wiersza-kontraktu w tabeli kontraktów możliwe było potwierdzenie lub anulowanie usunięcia kontraktu.

<tbody>
  {cursor && Object.keys(cursor).map(
      (keyName, keyIndex) =>
      <tr  className="has-text-centered is-size-7"
           key={keyIndex}>
          <td>{keyIndex+1}</td>
          <td>{(cursor[keyName].status)}</td>
          <td>{(cursor[keyName].contract_number)}</td>
          <td>{(cursor[keyName].contractor)}</td>
          <td>{(cursor[keyName].customer)}</td>
          <td>{(cursor[keyName].date_of_order)}</td>
          <td>{(cursor[keyName].date_of_delivery)}</td>
          <td>{(cursor[keyName].pallets_position)}</td>
          <td>{(cursor[keyName].pallets_planned)}</td>
          <td>{(cursor[keyName].pallets_actual)}</td>
          <td>{(cursor[keyName].warehouse)}</td>
          <td>
            <FaEdit />
            <FaTrashAlt onClick={() => 
               showModal(cursor[keyName].id)}
               title={`Delete contract ${keyIndex+1}`} />
          </td>
      </tr>)}   
</tbody>
Deleting contract

Zamiast bezpośrednio wywołać funkcję deleteContract() zostaje wywołana funkcja showModal() pobierająca jako parametr id wybranego kontraktu.

Funkcja showModal() ustawia odpowiednio zmienne id oraz modal tzn.

const [modal, setModal] = useState(false)
const [id, setId] = useState()

const showModal = (id) => {
        setId(id)
        setModal(!modal)
    }

Zmienna id przechowuje numer id wybranego kontraktu, natomiast zmienna modal przechowuje wartość boolean określającą stan okna modalnego (wyświetlony lub niewidoczny-domyślnie). Wywołanie funkcji setModal() zmienia wartość domyślną – false zmiennej modal na true.

Kod okna modalnego wyświetlającego potwierdzenie kasowania kontraktu wygląda następująco:

<div class={`modal ${modal && 'is-active'}`}>
  <div class="modal-background"></div>
  <div class="modal-card column is-one-quarter">
     <header class="modal-card-head">
        <p class="modal-card-title has-text-danger">
           Delete Contract?
        </p>
      </header>
      <section class="modal-card-foot">
         <button class="button is-danger" 
             onClick={() => deleteContract(id)}>Delete
         </button>
         <button class="button"
             onClick={() => setModal(!modal)}>Cancel
         </button>
       </section>
  </div>
</div>

Wybranie przycisku Cancel powoduje, że okno modalne staje się niewidoczne. Wybranie przycisku Delete aktywuje właściwą funkcję kasowania wiersza tzn.

const deleteContract = (id) => {
  setCursor(Object.values(cursor)
    .filter((row) => row.id !== id))
  fetch('/api/contract/delete', {
    method: 'POST',
    headers: {
      'Content-type': 'application/json',
     },
       body: JSON.stringify({'id': id, 'token': token})
  })
  .then(res => res.json())
  .then(data => console.log(data))
  .catch((err) => console.log(err));
  setModal(!modal)
}

Funkcja setCursor pobiera wartości obiektu cursor i filtruje je w taki sposób, że odrzucany jest wybrany kontrakt. Następnie następuje wywołanie api we Flasku, oprócz id kontraktu przesyłany jest również token użytkownika. Wynik pomyślnego kasowania wiersz wyświetlany jest w konsoli.

Jak wygląda API we Flasku opisałem w poprzednim wpisie.

W ostatnim wierszu funkcji deleteContract() usuwam okno modalne.

Kasowanie rekordu (Flask-RESTful, SQLAlchemy, React i Bulma CSS framework)

W tabeli kontraktów po naciśnięciu odpowiedniego przycisku kasowania wiersza w tabeli zostaje aktywowana funkcja deleteContract, która jako parametr przyjmuje wartość id kontraktu tzn.:

<td>
    <FaTrashAlt onClick={() => deleteContract(cursor[keyName].id)} />
</td>

Funkcja deleteContract filtruje dane zawarte w zmiennej cursor w taki sposób, że odrzucany jest wiersz, który chcemy usunąć. Następnie wywoływana jest zapytanie do backendu, gdzie w metodzie POST przesyłana jest wartość id kontraktu oraz token. Gdy usunięcie rekordu będzie zakończone sukcesem, to z API zostaje przesłana informacja w zmiennej message (w podanym przykładzie wartość tej zmiennej jest wyświetlana w konsoli).

const deleteContract = (id) => {
    setCursor(Object.values(cursor).filter((row) => row.id !== id))
    fetch('/api/contract/delete', {
            method: 'POST',
            headers: {
                'Content-type': 'application/json',
            },
            body: JSON.stringify({'id': id, 'token': token})
        })
        .then(res => res.json()).then(data => console.log(data)).catch((err) => console.log(err));
    }

Backend we Flask-RESTful z wykorzystaniem SQLAlchemy:

class DeleteContract(Resource):
    def post(self):
        req = request.json
        contract_id = req['id']
        token = req['token'].strip('"')
        user = User.query.filter_by(token=token).first()
        if user.role == 'Customer':
            contract = Contract.query.filter_by(customer=user.username).filter_by(id=contract_id).first()
            if contract:
                contract.status ='cancelled'
                db.session.commit()
                return {"message": "Row deleted"}

Przesłane za pomocą metody POST wartości id i token przypisuję do odpowiednich zmiennych contract_id oraz token. Następnie znajduję użytkownika w bazie danych na podstawie przesłanego tokena. Jeśli użytkownik należy do grupy Customer, to dokonywane jest wyszukiwanie kontraktu na podstawie nazwy użytkownika i numeru id kontraktu. Jeśli kontrakt ten istnieje to zamiast go usuwać z bazy, zmieniam jego status na cancelled.

Uaktualniłem również metodę post() klasy AllContracts(Resource), aby utworzony obiekt kursora nie zawierał rekordów ze statusem cancelled tzn.:

class AllContracts(Resource):
    def get(self, token):
        token = token.strip('"')
        user = User.query.filter_by(token=token).first()
        if user.role == 'Customer':
            contracts = Contract.query.filter(
                Contract.status!='cancelled').
                filter_by(customer=user.username).
                order_by(Contract.id.desc()).all()
        elif user.role == 'Contractor':
            contracts = Contract.query.
                    filter(Contract.status!='cancelled').
                    filter_by(contractor=user.username).
                    order_by(Contract.id.desc()).all()
        cursor = {}
        for i, contract in enumerate(contracts):
            cursor[i] = contract.serialize()
        return cursor

Hook useEffect() do wczytania kontraktów po wyborze opcji Contracts z menu wygląda następująco:

    useEffect( 
        () => {
            const token = sessionStorage.getItem('token')
            fetch('/api/contract/' + token)
            .then(res => res.json()).then(data => setCursor(data)).catch((err) => console.log(err))
        }, [token, setCursor],
    );

React, Flask-restful i SQLAlchemy – pobranie danych z bazy danych

Frontend:

Dane z backendu pobieram przy wybraniu z opcji menu komponentu Navbar frameworka Bulma. Aby dane były wczytane podczas randerowania komponentu używam hook useEffect. Tworzone jest wówczas zapytanie do backendu, a jako parametr przekazywany jest token, ustawiany podczas logowania użytkownika. Z backendu przesyłana jest odpowiedź w formie obiektu Promise, z którego jeśli zapytanie zostało zakończone sukcesem jest ustawiana zmienna cursor (zmiana stanu za pomocą setCursor). Cursor i funkcja setCursor przekazywane są do komponentu z komponentu nadrzędnego App.js jako parametry (props) tzn.

import LoginForm from '../Forms/LoginForm/LoginForm'
import { useEffect } from "react";

function Contracts({token, putToken, cursor, setCursor}) {

    useEffect( 
        () => {
            const token = sessionStorage.getItem('token')
            fetch('/api/contract/' + token)
            .then(res => res.json()).then(data => setCursor(data)).catch((err) => console.log(err))
        }, [token, setCursor],
    );
    
    if (!token) {
        return <LoginForm putToken={putToken} />
    };

Aby komponent odświeżał dane po poprawnym zalogowaniu się w dodatkowym parametrze metody useEffect w tablicy umieściłem zmienną token, która zostaje ustawiona w komponencie LoginForm.js

Aby komponent odświeżał dane po zmianie zawartości cursora np. poprzez dodanie nowego kontraktu jako dodatkowy parametr metody useEffect umieściłem funkcję setCursor.

Poszczególne wiersze danych są umieszczane w tabeli tzn.

<tbody>
    {cursor && Object.keys(cursor).map((keyName, keyIndex) => 
    <tr key={keyIndex}>
            <td>{keyIndex+1}</td>
            <td>{(cursor[keyName].status)}</td>
            <td>{(cursor[keyName].contract_number)}</td>
            <td>{(cursor[keyName].contractor)}</td>
            <td>{(cursor[keyName].customer)}</td>
            <td>{(cursor[keyName].date_of_order)}</td>
            <td>{(cursor[keyName].date_of_delivery)}</td>
            <td>{(cursor[keyName].pallets_position)}</td>
            <td>{(cursor[keyName].pallets_planned)}</td>
            <td>{(cursor[keyName].pallets_actual)}</td>
            <td>{(cursor[keyName].warehouse)}</td>
     </tr>)}   
</tbody>

Backend we Flask-restful:

W pliku __init__.py dodaję kolejną klasę zasobów:

from api.resources.contracts import AllContracts

api.add_resource(AllContracts, '/api/contract/<token>', endpoint='all_contracts')

W pliku contracts.py definiuję klasę zasobów:

from datetime import datetime
from flask import request
from flask_restful import Resource
from .. import db
from ..common.models import User, Contract

class AllContracts(Resource):
    def get(self, token):
        token = token.strip('"')
        user = User.query.filter_by(token=token).first()
        contracts = Contract.query.filter_by(customer=user.username).order_by(Contract.id.desc()).all()
        cursor = {}
        for i, contract in enumerate(contracts):
            cursor[i] = contract.serialize()
        return cursor

W pliku models.py definiuję m.in. jak klasa Contract dziedzicząca po klasie Model z SQLAlchemy ma być serializowana:

from .. import db
from datetime import datetime

class Contract(db.Model):
    '''Model of contract between a contractor and a customer'''

    id = db.Column(db.Integer, primary_key=True)
    status = db.Column(db.String(10), nullable=False, default='open')
    contract_number = db.Column(db.String(20), nullable=False)
    contractor = db.Column(db.String(20), db.ForeignKey('user.username'), nullable=False)
    customer = db.Column(db.String(20), db.ForeignKey('user.username'), nullable=False)
    date_of_order = db.Column(db.Date, nullable=False, default=datetime.utcnow)
    date_of_delivery = db.Column(db.Date, nullable=False)
    pallets_position = db.Column(db.Integer)
    pallets_planned = db.Column(db.Integer, nullable=False)
    pallets_actual = db.Column(db.Integer)    
    warehouse = db.Column(db.String(10), nullable=False)

    def serialize(self):
        return {'id': self.id,
                'status': self.status,
                'contract_number': self.contract_number,
                'contractor': self.contractor,
                'customer': self.customer,
                'date_of_order': datetime.strftime(self.date_of_order, '%Y-%m-%d'),
                'date_of_delivery': datetime.strftime(self.date_of_delivery, '%Y-%m-%d'),
                'pallets_position': self.pallets_position,
                'pallets_planned': self.pallets_planned,
                'pallets_actual': self.pallets_actual,
                'warehouse': self.warehouse 
                }

React i Bulma navbar

Rozwiązane problemy:

  1. Jak zwinąć burger menu przy wyborze opcji z w/w menu. (domyślnie wybór opcji z burger menu pozostawia rozwinięte menu)
  2. Jak zwinąć menu rozwijalne przy wyborze którejś z opcji. (domyślnie menu rozwijalne nie zamyka się, gdy wybrano którąś z opcji)

ad. 1

Menu nareszcie zwija się po wybraniu opcji
<div id="navbarBasicExample" className={`navbar-menu ${burgerActive? "is-active": ""}`}>
                <div class="navbar-start">
                <Link to="/" class="navbar-item" onClick={handleOnClick}>
                    Home
                </Link>

przy czym:

    const [burgerActive, setBurgerActive] = useState(false)

    const handleOnClick = () => {
        setBurgerActive(false)
    }

ad2.

Menu rozwijalne zwija się, gdy wybrano którąś z opcji.
<div class="navbar-item has-dropdown is-hoverable" key={location.pathname}>

przy czym:

import {
    Link, useLocation
  } from 'react-router-dom' 

const Navbar = () => {
    let location = useLocation();

Obsługa zamówień dla Centrów logistycznych

Aplikacja umożliwia wystawianie nowych zamówień przez Odbiorcę oraz akceptację zamówień przez Dostawcę i generowanie rezerwacji tzn. zarządzanie oknami czasowymi z myślą o dostawach do magazynów centralnych i centrów logistycznych.

Kod źródłowy aplikacji do pobrania – tutaj.

W tym wpisie przedstawię funkcję new_booking(id), która obsługuje rezerwację terminu dla konkretnego zamówienia. Jako parametr przyjmuje ona numer id zamówienia tzn.

@bp.route('/booking/<int:id>', methods=['GET', 'POST'])
@login_required
def new_booking(id):

Następnie w funkcji new_booking() tworzę formularz do rezerwacji oraz pobieram obiekt rezerwacji dla danego zamówienia tzn.

    form = BookingForm()
    result = Booking.query.filter_by(contract_id=id).first()

Dalej ustalam obiekt zamówienia dla rezerwacji tzn.

    current_contract = Contract.query.get(id)     # Returns contract for current booking

W kolejnych wierszach uzyskuję listę zamówień, które były wygenerowane dla konkretnego magazynu na dzień dostawy tzn.

# Filtering by one column gives a list of tuple(s) so I converted it to a list of values
    contracts = [ids[0] for ids in Contract.query.with_entities(Contract.id).filter_by(
        date_of_delivery=current_contract.date_of_delivery).filter_by(
        warehouse=current_contract.warehouse).all()]

Następnie przeprowadzam walidację, jeśli formularz został przesłany. W zależności od tego, czy rezerwacja dla danego zamówienia jest już utworzona to uaktualniam dane w bazie danych na podstawie danych z formularza. Jeśli obiekt rezerwacji nie istnieje to tworzę nowy obiekt rezerwacji i zapisuję go do bazy danych tzn.

    if form.validate_on_submit():
        if result:
            result.booking_time = form.booking_time.data
            result.driver_full_name = form.driver_full_name.data
            result.driver_phone_number = form.driver_phone_number.data
            result.truck_reg_number = form.truck_reg_number.data
            db.session.commit()
        else:
            booking = Booking(booking_time=form.booking_time.data, 
                            contract_id = id,
                            driver_full_name=form.driver_full_name.data,
                            driver_phone_number=form.driver_phone_number.data,
                            truck_reg_number=form.truck_reg_number.data)
            db.session.add(booking)
            db.session.commit()

W dalszej kolejności zmieniam status zamówienia na zaakceptowane i przekierowuję do funkcji wyświetlającej wszystkie zamówienia dla danego kontrahenta tzn.

        contract = Contract.query.get(id)
        contract.status = 'accepted'
        db.session.commit()
        page = session.get('page')
        per_page = session.get('per_page')
        return redirect(url_for('contracts.contracts', page=page, per_page=per_page))

Jeśli strona zostaje wczytana za pomocą metody GET to w zależności czy rezerwacja jest już obecna w systemie, czy dopiero jest tworzona to uzupełniane są dane w formularzu tzn. jeśli rezerwacja jest dostępna i otwieramy ją np. w celu edycji danych to formularz zostanie uzupełniony o poprzednio wpisane dane. Natomiast jeśli rezerwacja jest dopiero tworzona to formularz będzie zawierał modyfikację, których terminów nie można wybrać, bo już zostały wybrane przez innych dostawców tzn.

    if result is not None:
        reserved_booking_time = [times[0] for times in 
            Booking.query.with_entities(Booking.booking_time).filter(
            Booking.contract_id.in_(contracts)).all() if times[0]!=result.booking_time]   
        form.booking_time.data = result.booking_time
        form.driver_full_name.data = result.driver_full_name
        form.driver_phone_number.data = result.driver_phone_number
        form.truck_reg_number.data = result.truck_reg_number   
    else:
        reserved_booking_time = [times[0] for times in 
            Booking.query.with_entities(Booking.booking_time).filter(
            Booking.contract_id.in_(contracts)).all()] 
    return render_template('booking.html', form=form, reserved_booking_time=reserved_booking_time)

Kompletny kod funkcji new_booking() jest dostępny tutaj.

Tak zaakceptowane zamówienie może być edytowane tylko przez wystawiającego zamówienie, przy czym jeśli zamówienie zostało już zaakceptowane i została dokonana rezerwacja, to zmiana zamówienia zmienia status zamówienia z zaakceptowanego z powrotem na otwarty i wymaga powtórnego zaakceptowania zmian przez dostawcę (ale wpisane dane są obecne i rezerwacja nie wymaga powtórnego uzupełnienia danych).

Dostawca może również pobrać dokument pdf rezerwacji jeśli zamówienie jest zaakceptowane (dla otwartych zamówień opcja pobrania pdf-a jest nieaktywna).

Rezerwacja dokonana – można teraz wygenerować pdf-a

Reportlab – wygenerowanie i pobranie pdf-a we Flasku bez tworzenia pliku na serwerze

Rozwiązany problem: jak wygenerować pdf z danymi zamówienia i kodem paskowym rezerwacji bez zapisu pliku na dysku, który może być następnie pobrany przez użytkownika serwisu we Flasku.

Do osadzenia elementów dokumentu takich jak teksty, tabele itp. możemy użyć obiekt Canvas. Jako argument możemy podać nazwę, jaką będzie posiadał utworzony pdf, ale w tym przypadku zamiast tworzyć plik na serwerze użyję bufora. Jawnie podaję rozmiar strony jako A4.

from io import BytesIO
from reportlab.pdfgen.canvas import Canvas
from reportlab.lib.pagesizes import A4

buffer = BytesIO()
canvas = Canvas(buffer, pagesize=A4)

Aby użyć niestandardowych znaków np. polskich liter z ogonkami należy zarejestrować odpowiednie czcionki, które obsługują wymagane znaki. W przykładowym pliku pdf użyłem czcionki Vera.

from reportlab.pdfbase import pdfmetrics
from reportlab.pdfbase.ttfonts import TTFont

pdfmetrics.registerFont(TTFont('Vera', 'Vera.ttf'))

Tak zarejestrowaną czcionkę można już użyć:

canvas.setFont("Vera", size=10)

Obrazy wygenerowane na podstawie danych z aplikacji (w tym przypadku kod kreskowy rezerwacji) umieszczam w tworzonym dokumencie za pomocą metody drawImage(), która jako pierwszy argument pobiera obiekt ImageReader tzn.

im = ImageReader(image)
canvas.drawImage(im, x=0, y=-5*cm, width=150, height=100)

Listę zawierającą łańcuchy tekstowe można dodać do dokumentu wykorzystując obiekt tekstowy tzn.

txt_obj = canvas.beginText(14, -6.5 * cm)
txt_lst = ["line of text 1", "line of text 2", "line of text 3"]
for line in txt_lst:
        txt_obj.textOut(line)
        txt_obj.moveCursor(0, 16)
canvas.drawText(txt_obj)

Tabelę do dokumentu można dodać za pomocą obiektu klasy Table tzn. osobno definiuję listę zawierającą dane poszczególnych wierszy tabeli (zmienna table_data). Osobno również definiuję style obowiązujące w całej lub w części tabeli. 

t = Table(table_data, colWidths=[60, 230, 70, 60, 50], rowHeights=30)
style = [('BACKGROUND',(0,0),(-1,-2),colors.lightblue),
            ('ALIGN',(0,-1),(-1,-1),'CENTER'),
            ('BOX', (0,0), (-1,-1), 0.25, colors.black),
            ('INNERGRID', (0,0), (-1,-1), 0.25, colors.black),
            ('FONTSIZE', (0,0), (-1,-1), 10),
            ('ALIGN', (0, 0), (-1, -1), 'CENTER'),
            ('VALIGN', (0, 0), (-1, -1), 'MIDDLE')]
t.setStyle(tblstyle=style)
t.wrapOn(canvas, 10, 10)
    t.drawOn(canvas=canvas, x=20, y=-22*cm)

Następnie zapisuję tak utworzony obiekt Canvas i zwracam powstały bufor do kontrolera. W celu przesłania pliku pdf na podstawie danych z bufora wykorzystuję funkcję send_file() Flaska:

from flask.helpers import send_file

@bp.route('/get-pdf/<int:id>', methods=['GET'])
@login_required
def get_pdf(id):
    contract = Contract.query.get(id)
    contractor = User.query.get(contract.contractor_id)
    booking = Booking.query.filter_by(contract_id=contract.id).first()
    pdf = create_pdf(booking_no=booking.id,
                        contractor=contractor.username,
                        contractor_no=contractor.id,
                        truck_plate=booking.truck_reg_number,
                        warehouse=contract.warehouse,
                        date=contract.date_of_delivery,
                        time=booking.booking_time,
                        pallets_pos=contract.pallets_position,
                        pallets=contract.pallets_actual)
    pdf.seek(0)
    return send_file(pdf, as_attachment=True, mimetype='application/pdf',
        attachment_filename='booking.pdf', cache_timeout=0)
przykładowy plik wygenerowany po stronie serwera, pobrany przez zleceniobiorcę.

System zarządzania oknami czasowymi we Flasku #2

Kod źródłowy aplikacji do pobrania – tutaj.

Plan aplikacji opisanej we wcześniejszym artykule wygląda następująco:

├───flask_contracts
│ ├───api
│ ├───auth
│ │ ├───static
│ │ │ └───logos
│ │ ├───templates
│ │ │ └───auth
│ │ ├───utils
│ │ │ ├───templates
│ │ │ │ └───utils
│ ├───contracts
│ │ ├───static
│ │ ├───templates
│ │ │ └───contracts
│ │ │ └───errors
│ │ ├───utils
│ ├───static
│ ├───templates
├───tests
│ ├───functional
│ ├───unit
└───venv

Każda część aplikacji jest osobną całością. Podobne rozwiązanie jest zastosowane w Django. Do zarządzania aplikacją napisałem plik manage.py. Każda część składowa posiada własne podkatalogi szablonów – konwencja podobna do obecnej w Django.

System zarządzania oknami czasowymi za pomocą Flask

Dzięki aplikacji: minimalizuje się przestoje oraz optymalizuje się pracę magazynu, co wpłynie na redukcję kosztów i brak przestojów.

Aplikacja umożliwia zarządzanie oknami czasowymi z myślą o dostawach do magazynów centralnych i centrów logistycznych.

Kod źródłowy aplikacji do pobrania – tutaj.

Projekt będzie zawierał aplikację we Flasku (z użyciem frameworka css – Bulma).

Do aplikacji będą mogli się logować dostawcy i odbiorcy dostaw (weryfikacja konta mailem na podany adres) .

Zlecenia będą mogli wystawiać tylko odbiorcy dostaw, natomiast dostawcy będą mogli samodzielnie rezerwować dostępne okna czasowe i wygenerować dokument dostawy z kodem paskowym jako plik pdf.

— część 2 —

Przekazanie wartości do i z modala (framework Bulma) we Flasku

Rozwiązany problem: jak przekazać wartość do okna modalnego oraz jak przekazać wartość z okna modalnego do aplikacji we Flasku.

W przedstawianym przypadku każdy wiersz tabeli html zawiera przycisk umożliwiający anulowanie kontraktu. Dane dla poszczególnych wierszy są przekazywane z kontrolera za pomocą zmiennej contracts:

result = Contract.query.filter_by(
            customer_id=current_user.id).filter(
                Contract.status!='cancelled').order_by(
                    Contract.id.desc()).paginate(page=page, per_page=per_page)
return render_template('contracts.html', title='Contracts', header=columns, contracts=result)

Dane dla wierszy są dostępne w szablonie:

{% for contract in contracts.items %}
    <tr>
        <td>{{  loop.index  }}</td>
        {% for item in header %}
            <td>
                 {{ contract[item] }}
            </td>
        {% endfor %}
    </tr>
{% endfor %}

Oprócz danych przekazanych z kontrolera należy dla każdego wiersza tabeli utworzyć przycisk uruchamiający odpowiedni modal – różniący się wartością parametru data-target tzn.

<button class="button mr-2 modal-button" data-target="modal{{contract.id}}"  title="Cancel Contract">

Wartość parametru data-target będzie się różnić dla każdego wiersza w zależności od wartości contract.id , która to wartość jest unikatowa – klucz tabeli Contract.

W zależności od wartości parametru data-target uruchamiany jest modal o takiej samej wartości parametru id, tzn.

<div id="modal{{contract.id}}" class="modal">

Modal zawiera przycisk do zatwierdzenia zmian, w tym przypadku do anulowania kontraktu. Aby przekazać, jaki kontrakt ma zostać anulowany należy podać id kontraktu jako parametr do funkcji url_for() , tzn.

<form action="{{ url_for('contracts.cancel_contract', id=contract.id) }}">
            <button class="button is-danger is-rounded" formmethod="POST">Yes</button>
</form>