W dzisiejszej części, połączymy nasz serwer napisany w Flasku z użyciem Celery, z aplikacją działającą po stronie backendu. Docelowo w trzeciej częsci zaprzęgniemy nasz serwer domowy z zainstalowaną aplikacją napisaną w Flask-u do pracy oraz pozwolimy naszemu serwerowi działającemu na vps-ie u klienta odetchnąć i zoptymalizować koszty.
Na samym początku zakładamy że utworzyliśmy już domyślną aplikację Django zgodnie np. z oficjalnym tutorialem.
python3 -m venv venv
source venv/bin/activate
pip install -r requirements.txt
django-admin startproject finder
A więc zaczynamy :)
1.Modele
Na początku tworzymy nasz pierwszy model, pozwalający przechować otrzymane dane, zapisując je w bazie danyh.
Każdy rekord będzie się składał z:
- guid - zapisany będzie w nim krótki identyfikator tekstowy naszego zadania,
- autor - przypiszemy do niego osobe która doda wymienione zadanie,
- query - przechowywać będzie nasze zapytanie które prześlemy do naszego api.
- result - cały wynik poszukiwań zostanie w nim zapisany,
- status - przechowa nie tylko aktualny status ale i również komunikaty o błędach,
- created date - zapiszemy w nim datę dodania zadania (będzie ustawiana automatycznie),
Czas na kod:
# models.py
from datetime import datetime
from django.contrib.auth.models import User
from django.db import models
from django.utils.translation import ugettext_lazy as _
class BigFilter(models.Model):
guid = models.TextField(_('Guid'), max_length=100, null=True)
author = models.ForeignKey(User, on_delete=models.CASCADE, verbose_name=_('Autor'), null=True)
query = models.CharField(_('Query'), max_length=100, null=True)
result = models.TextField(_('Result'), max_length=3000, null=True)
status = models.CharField(_('Status'), max_length=100, null=True)
created_date = models.DateTimeField(_('Date created'), default=datetime.now, null=True)
def __str__(self):
return self.query
class Meta:
verbose_name = _('Result')
verbose_name_plural = _('Search engine')
2. Panel administracyjny
Całą operację zarządzania zadaniami, tworzenia ich jak i zarządzania poziomem dostępu uprawnień, przekażemy panelu administacyjnemu.
Nasz kod będzie składał się z:
Metoda save_model
W samym kodzie podczas zapisywania naszego modelu, wykonujemy zapytanie do api Flaskowego, które następnie za pomocą Celery, rozpocznie poszukiwania interesującego nas ciągu znaków. W przypadku jakich kolwiek problemów otrzymamy informację zwrotną z błędem.
def save_model(self, request, obj, form, change):
if not change:
obj.author = request.user
try:
url = "{}/find".format(settings.FINDER_SERVER)
payload = {'data': obj.query}
headers = {"User-Agent": "curl/7.61.0"}
response = requests.request("POST", url, headers=headers, data=payload)
obj.guid = response.json()['result_id']
messages.add_message(request, messages.INFO, 'Success: guid {}'.format(obj.guid))
except Exception as e:
messages.add_message(request, messages.INFO, 'Error: {}'.format(e))
super().save_model(request, obj, form, change)
Metoda result_field
Na liście wszystkich naszych dodanych zadań, zdefiniowaliśmy własne pole "result_field". Które po przez to że przy każdym załadowaniu listy będzie miało (po przez zmienną obj) dostęp do danych, będzie potrafiło sprawdzić czy dla danego pola otrzymaliśmy wynik, jeśli tak to załaduje jego wynik. Jeśli nie to pobierze aktualny status oraz wynik polecenia.
def result_field(self, obj):
if obj.status == "Completed":
return obj.status
url = "{}/result/{}".format(settings.FINDER_SERVER, obj.guid)
headers = {'Content-Type': 'application/json'}
response = requests.request("POST", url, headers=headers)
response.encoding = 'UTF-8'
if requests.status_code != 200:
raise Exception("Api error: {}/{} ".format(requests.status_code, requests.reason))
obj = BigFilter.objects.get(id=obj.id)
if response.json()['ready'] and response.json()['successful']:
obj.status = "Completed"
else:
obj.status = response.json()
try:
obj.result = ""
for data in json.loads(response.json()['value']):
obj.result += "{}:{} {}\n".format(data['file'], data['line'], data['data'])
except:
obj.result = response.json()['value']
obj.save()
return obj.status
result_field.short_description = 'Status'
Aby zachować całe flow (gdzie zapytania do api wykonywane są tylko na liście), nadpisaliśmy metody zapisu oraz zmiany danych tak, by przekierowywały na listę i tym samym pozwalały powyższej metodzie działać.
def response_add(self, request, obj, post_url_continue=None):
return redirect('/admin/finder/bigfilter')
def response_change(self, request, obj):
return redirect('/admin/finder/bigfilter')
Cały kod:
import json
import requests
from django.conf import settings
from django.contrib import admin, messages
from django.shortcuts import redirect
from models import BigFilter
class BigFilterAdmin(admin.ModelAdmin):
list_display = ('id', 'autor_name', 'query', 'result_field', 'created_date')
readonly_fields = ('created_date', 'author', 'status', 'result')
exclude = ('guid',)
def autor_name(self, obj):
return obj.author.get_full_name()
autor_name.short_description = 'Creator'
def result_field(self, obj):
if obj.status == "Completed":
return obj.status
url = "{}/result/{}".format(settings.FINDER_SERVER, obj.guid)
headers = {'Content-Type': 'application/json'}
response = requests.request("POST", url, headers=headers)
response.encoding = 'UTF-8'
if requests.status_code != 200:
raise Exception("Api error: {}/{} ".format(requests.status_code, requests.reason))
obj = BigFilter.objects.get(id=obj.id)
if response.json()['ready'] and response.json()['successful']:
obj.status = "Completed"
else:
obj.status = response.json()
try:
obj.result = ""
for data in json.loads(response.json()['value']):
obj.result += "{}:{} {}\n".format(data['file'], data['line'], data['data'])
except:
obj.result = response.json()['value']
obj.save()
return obj.status
result_field.short_description = 'Status'
def save_model(self, request, obj, form, change):
if not change:
obj.author = request.user
try:
url = "{}/find".format(settings.FINDER_SERVER)
payload = {'data': obj.query}
headers = {"User-Agent": "curl/7.61.0"}
response = requests.request("POST", url, headers=headers, data=payload)
obj.guid = response.json()['result_id']
messages.add_message(request, messages.INFO, 'Success: guid {}'.format(obj.guid))
except Exception as e:
messages.add_message(request, messages.INFO, 'Error: {}'.format(e))
super().save_model(request, obj, form, change)
def response_add(self, request, obj, post_url_continue=None):
return redirect('/admin/finder/bigfilter')
def response_change(self, request, obj):
return redirect('/admin/finder/bigfilter')
admin.site.register(BigFilter, BigFilterAdmin)
W trzeciej części umożliwimy naszej aplikacji Flaskowej, znajdującej się na komputerze w domu, dostęp do świata (serwera).
Komentarze