Python库 | wagtail-airtable-0.1.0.tar.gz

import sys from importlib import import_module from airtable import Airtable from django.db import models, IntegrityError from django.conf import settings from django.contrib.contenttypes.models import ContentType from django.core.exceptions import ObjectDoesNotExist, ValidationError from django.core.management.base import BaseCommand from logging import getLogger from modelcluster.contrib.taggit import ClusterTaggableManager from taggit.managers import TaggableManager from wagtail.core.models import Page from wagtail_airtable.tests import MockAirtable from wagtail_airtable.utils import get_model_for_path, get_validated_models logger = getLogger(__name__) DEFAULT_OPTIONS = { "verbosity": 1, } TESTING = any(x in ["test", "runtests.py"] for x in sys.argv) class Importer: def __init__(self, models=[], options=DEFAULT_OPTIONS): self.models = models self.options = options self.records_used = [] self.cached_records = {} self.created = 0 self.updated = 0 self.skipped = 0 def debug_message(self, message): """ Local function. Print debug messages if `verbosity` is 2 or higher. """ if self.options["verbosity"] >= 2: if not TESTING: print(message) return message def get_model_serializer(self, serializer_string): location, serializer_name = serializer_string.rsplit(".", 1) module = import_module(location) serializer_class = getattr(module, serializer_name) return serializer_class def get_model_settings(self, model) -> dict: return settings.AIRTABLE_IMPORT_SETTINGS.get(model._meta.label, {}) def get_column_to_field_names(self, airtable_unique_identifier) -> tuple: uniq_id_type = type(airtable_unique_identifier) airtable_unique_identifier_column_name = None airtable_unique_identifier_field_name = None if uniq_id_type == str: # The unique identifier is a string. # Use it as the Airtable Column name and the Django field name airtable_unique_identifier_column_name = airtable_unique_identifier airtable_unique_identifier_field_name = airtable_unique_identifier elif uniq_id_type == dict: # Unique identifier is a dictionary. # Use the key as the Airtable Column name and the value as the Django Field name. ( airtable_unique_identifier_column_name, airtable_unique_identifier_field_name, ) = list(airtable_unique_identifier.items())[0] return ( airtable_unique_identifier_column_name, airtable_unique_identifier_field_name, ) def get_or_set_cached_records(self, airtable_client): # Memoize results from Airtable so we don't hit the same API multiple times # This is largely used to support additional Wagtail/Airatble settings # that are identical to each other, as in they would use the # same Airtable Base as the key in the dictionary. # ie. # 'yourapp.YourPage': { # ... # 'AIRTABLE_TABLE_NAME': 'Your Table', # }, # 'different_app.DifferentPage': { # ... # 'AIRTABLE_TABLE_NAME': 'Your Table', # Same Airtable Table name # } # All of the above settings will use the 'Your Table' results # instead of hitting the Airtable API for each model and getting the same # results every time. This is designed to help with API efficiency, reduce # load/import times, and to reduce how much memory is required to save all # the records from Airtable. if self.cached_records.get(airtable_client.table_name): all_records = self.cached_records.get(airtable_client.table_name) else: # Get all the airtable records for the specified table. all_records = airtable_client.get_all() self.cached_records[airtable_client.table_name] = all_records return all_records def convert_mapped_fields(self, record_fields_dict, mapped_fields_dict) -> dict: # Create a dictionary of newly mapped key:value pairs based on the `mappings` dict above. # This wil convert "airtable column name" to "django_field_name" mapped_fields_dict = { mapped_fields_dict[key]: value for (key, value) in record_fields_dict.items() if key in mapped_fields_dict } return mapped_fields_dict def update_object( self, instance, record_id, serialized_data, is_wagtail_model=False ) -> bool: """ Attempts to update an object. Returns a bool that determines if the object was updated or not. """ if serialized_data.is_valid(): self.debug_message( "\t\t Serializer data was valid. Setting attrs on model..." ) model = type(instance) for field_name, value in serialized_data.validated_data.items(): field_type = type( model._meta.get_field(field_name) ) # ie. django.db.models.fields.CharField # If this field type is a subclass of a known Wagtail Tag, or a Django m2m field # We need to loop through all the values and add them to the m2m-style field. if issubclass( field_type, (TaggableManager, ClusterTaggableManager, models.ManyToManyField,), ): m2m_field = getattr(instance, field_name) for m2m_value in value: m2m_field.add(m2m_value) else: setattr(instance, field_name, value) # When an object is saved it should NOT push its newly saved data back to Airtable. # This could theoretically cause a loop. By default this setting is True. But the # below line confirms it's false, just to be safe. instance.airtable_record_id = record_id instance.push_to_airtable = False try: if is_wagtail_model: self.debug_message("\t\t This is a Wagtail Page model") # Wagtail page. Requires a .save_revision() if not instance.locked: self.debug_message( "\t\t\t Page is not locked. Saving page and creating a new revision." ) # Only save the page if the page is not locked instance.save() instance.save_revision() self.updated = self.updated + 1 else: self.debug_message("\t\t\t Page IS locked. Skipping Page save.") self.skipped = self.skipped + 1 else: # Django model. Save normally. self.debug_message("\t\t Saving Django model") instance.save() self.updated = self.updated + 1 # New record being processed. Save it to the list of records. self.records_used.append(record_id) # Object updated (and record was used) return True except ValidationError as error: self.skipped = self.skipped + 1 error_message = "; ".join(error.messages) logger.error( f"Unable to save {instance._meta.label} -> '{instance}'. Error(s): {error_message}" ) self.debug_message( f"\t\t Could not save Wagtail/Django model. Error: {error_message}" ) else: logger.info(f"Invalid data for record {record_id}")