import logging
from pathlib import Path
from django.core.files import File
from ferrosoft.apps.dataimport import queue
from ferrosoft.apps.dataimport.models import (
ImportJob,
MappingCollection,
FormatParameters,
DateFormatType,
Processors,
)
from ferrosoft.apps.ferrobase.management.tenantcommand import TenantAwareCommand
from ferrosoft.apps.ferrobase.models import User
[docs]
class Command(TenantAwareCommand):
[docs]
def add_arguments(self, parser):
parser.add_argument(
"-c",
"--collection",
help="Mapping collection name",
type=str,
required=True,
)
parser.add_argument(
"-e",
"--encoding",
help="Text encoding of import file",
default="",
type=str,
)
parser.add_argument(
"-O",
"--owner",
help="User name (will be owner of the file)",
type=str,
required=True,
)
parser.add_argument(
"--fieldsep",
help="CSV field separator",
type=str,
default=ImportJob.DEFAULT_FIELD_SEPARATOR,
)
parser.add_argument(
"--decimalsep",
help="Decimal separator",
type=str,
default=ImportJob.DEFAULT_DECIMAL_SEPARATOR,
)
parser.add_argument(
"--dateformat",
help="Date format type",
type=str,
default=DateFormatType.ISO_8601.value,
)
parser.add_argument(
"--datecodes",
help="Date format codes",
type=str,
default=ImportJob.DEFAULT_DATE_FORMAT_CODES,
)
parser.add_argument(
"--datetimecodes",
help="Date time format codes",
type=str,
default=ImportJob.DEFAULT_DATE_TIME_FORMAT_CODES,
)
parser.add_argument(
"--beforejob",
help="Function to run before job",
type=str,
default="",
)
parser.add_argument(
"--afterjob",
help="Function to run after job",
type=str,
default="",
)
parser.add_argument(
"--onmodel",
help="Function to run for each model",
type=str,
default="",
)
parser.add_argument(
"file",
help="Import file",
type=str,
)
[docs]
def handle_tenant(self, *args, **options):
logging.basicConfig(level=logging.INFO)
owner = User.get_by_username(options["owner"]).local_user()
collection = MappingCollection.objects.get(name=options["collection"])
file_path = Path(options["file"])
with open(options["file"], "rb") as import_file:
job = ImportJob.create(
File(import_file, file_path.name),
file_path.name,
options["encoding"] or None,
owner,
collection,
FormatParameters(
options["fieldsep"],
options["decimalsep"],
DateFormatType(options["dateformat"]),
options["datecodes"],
options["datetimecodes"],
),
Processors(
before_job=options["beforejob"] or None,
after_job=options["afterjob"] or None,
model=options["onmodel"] or None,
),
)
queue.process_import_job(job.pk)