Hier is een aangepaste Django REST Framework-client die offers kopieert via een API in plaats van direct in de database. Het gebruikt requests om de API-oproepen uit te voeren en stuurt de offers in JSON-formaat naar een andere server:
import requests
from django.conf import settings
from offers.models import Offer # Zorg dat je Offer-model correct is ingesteld in je Django-app.
class OfferSyncClient:
def __init__(self, api_base_url, api_token):
"""
Initialiseer de client met de API basis-URL en authenticatietoken.
"""
self.api_base_url = api_base_url
self.headers = {
"Authorization": f"Token {api_token}",
"Content-Type": "application/json",
}
def sync_offers(self, contact_types, post_codes):
"""
Kopieer de offers gebaseerd op contact types en postcodes via API.
"""
for ct in contact_types:
self._sync_offers_by_type(ct, post_codes)
def _sync_offers_by_type(self, contact_type, post_codes):
"""
Kopieer offers voor een specifieke contact type.
"""
offers = Offer.objects.filter(contact_type=contact_type, post_code__in=post_codes, section="sale")
print(f"Syncing {offers.count()} offers for contact type: {contact_type}")
for offer in offers:
self._send_offer_to_api(offer)
def _send_offer_to_api(self, offer):
"""
Stuur een enkele offer naar de API.
"""
payload = {
"slug": offer.slug,
"property_id": offer.property_id,
"title": offer.title,
"realty_type": offer.realty_type,
"section": offer.section,
"post_code": offer.post_code,
"location": offer.location,
"address": offer.address,
"price": str(offer.price) if offer.price else None,
"total_surface": offer.total_surface,
"livable_surface": offer.livable_surface,
"short_description": offer.short_description,
"number_of_bedrooms": offer.number_of_bedrooms,
"bathrooms": offer.bathrooms,
"furnished": offer.furnished,
"type_of_construction": offer.type_of_construction,
"monthly_costs": offer.monthly_costs,
"for_index": offer.for_index,
"hash": offer.hash,
"property_url": offer.property_url,
"created": offer.created.isoformat() if offer.created else None,
"last_status_update": offer.last_status_update.isoformat() if offer.last_status_update else None,
"contact_type": offer.contact_type,
"status": offer.status,
"phone": offer.contact.phone if offer.contact else None,
"mobile": offer.contact.mobile if offer.contact else None,
"contacted": offer.contacted,
"contacted_date": offer.contacted_date.isoformat() if offer.contacted_date else None,
"nota": offer.nota,
"image_url": offer.main_image,
"deal_stage": offer.deal_stage,
"user": offer.user.id if offer.user else None,
}
try:
response = requests.post(f"{self.api_base_url}/offers/", json=payload, headers=self.headers)
response.raise_for_status()
print(f"Offer {offer.id} successfully synced.")
except requests.exceptions.RequestException as e:
print(f"Failed to sync offer {offer.id}: {e}")
# Gebruik de client in een Django management command of service
from django.core.management.base import BaseCommand
from leads.models import LeadsConfig
class Command(BaseCommand):
help = "Sync offers with external API"
def handle(self, *args, **options):
config = LeadsConfig.objects.first()
if not config:
self.stdout.write("No configuration found.")
return
post_codes = [int(code.strip()) for code in config.postcodes.split(',')]
contact_types = [ct.strip() for ct in config.contact_type.split(',')]
client = OfferSyncClient(api_base_url=settings.EXTERNAL_API_BASE_URL, api_token=settings.EXTERNAL_API_TOKEN)
client.sync_offers(contact_types, post_codes)
To manage API tokens for Django Rest Framework (DRF) users through the Django admin interface, follow these steps:
Step 1: Enable the authtoken app
Ensure that rest_framework.authtoken is included in your INSTALLED_APPS:
INSTALLED_APPS = [
...
'rest_framework',
'rest_framework.authtoken',
]
python manage.py migrate
Token model in the Django Admin is already registered and you will see model in admin to create a Token by user.
# This is only for API
class OfferViewSet(viewsets.ModelViewSet):
queryset = Offer.objects.all()
serializer_class = OfferSerializer
#permission_classes = [permissions.IsAuthenticated]
def is_staff_and_can_update_offer(user):
"""
Custom test function to check if a user is both a staff member and has
the 'change_bauwensoffer' permission in the bauwens.models application.
"""
# Check if the user is a staff member
if user.is_staff:
return True
# Check if the user has the permission to change BauwensOffer objects
if user.has_perm('leads.change_offer'):
return True
# has the permission in group
if 'leads.change_offer' in user.get_group_permissions():
return True
return False
In setting
REST_FRAMEWORK = {
# Use Django's standard `django.contrib.auth` permissions
# Authenticated users require specific model-level permissions
# Unauthenticated users are denied access
'DEFAULT_PERMISSION_CLASSES': [
'rest_framework.permissions.IsAuthenticated',
],
'DEFAULT_AUTHENTICATION_CLASSES': [
'rest_framework.authentication.TokenAuthentication',
],
'DEFAULT_PAGINATION_CLASS': 'rest_framework.pagination.PageNumberPagination',
'PAGE_SIZE': 100,
}
In urls you need something like this.
urlpatterns = [
path('admin/login/', CustomAdminLoginView.as_view(), name='admin-login'),
path('', IndexView.as_view(), name='index'),
path('admin/', admin.site.urls),
path('lead/', include('lead.urls')),
path('api-auth/', include('rest_framework.urls')),
path('api/', include(router.urls)),
] + static(settings.MEDIA_URL, document_root=settings.MEDIA_ROOT)
Comments
Post a Comment