Prepare ncf manager for Tryton 8.0
This commit is contained in:
José Arturo García
@@ -0,0 +1,248 @@
|
||||
# This file is part of Tryton. The COPYRIGHT file at the top level of
|
||||
# this repository contains the full copyright notices and license terms.
|
||||
from decimal import Decimal
|
||||
|
||||
from sql import Literal, Null
|
||||
from sql.aggregate import Sum
|
||||
from sql.conditionals import Coalesce
|
||||
|
||||
from trytond import backend
|
||||
from trytond.i18n import gettext
|
||||
from trytond.model import ModelSQL, ValueMixin, fields
|
||||
from trytond.modules.company.model import (
|
||||
CompanyMultiValueMixin, CompanyValueMixin)
|
||||
from trytond.modules.currency.fields import Monetary
|
||||
from trytond.pool import Pool, PoolMeta
|
||||
from trytond.pyson import Eval, If
|
||||
from trytond.tools import grouped_slice, reduce_ids
|
||||
from trytond.transaction import Transaction
|
||||
from trytond.modules.ncf_manager.ncf_manager import NCF_TYPE
|
||||
import logging
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
def _backend_name():
|
||||
name = backend.name
|
||||
return name() if callable(name) else name
|
||||
|
||||
|
||||
customer_ncf_type = fields.Selection(
|
||||
NCF_TYPE,
|
||||
"Customer NCF Type",
|
||||
sort=False
|
||||
)
|
||||
|
||||
supplier_ncf_type = fields.Selection(
|
||||
NCF_TYPE,
|
||||
"Supplier NCF Type",
|
||||
sort=False
|
||||
)
|
||||
|
||||
|
||||
class Party(metaclass=PoolMeta):
|
||||
__name__ = 'party.party'
|
||||
customer_ncf_type = fields.MultiValue(customer_ncf_type)
|
||||
supplier_ncf_type = fields.MultiValue(supplier_ncf_type)
|
||||
|
||||
@classmethod
|
||||
def multivalue_model(cls, field):
|
||||
pool = Pool()
|
||||
if field in {'customer_ncf_type', 'supplier_ncf_type'}:
|
||||
return pool.get('party.party.ncf_type')
|
||||
return super(Party, cls).multivalue_model(field)
|
||||
|
||||
@classmethod
|
||||
def default_customer_ncf_type(cls, **pattern):
|
||||
pool = Pool()
|
||||
Configuration = pool.get('account.configuration')
|
||||
config = Configuration(1)
|
||||
return config.get_multivalue(
|
||||
'default_customer_ncf_type', **pattern)
|
||||
|
||||
@classmethod
|
||||
def default_supplier_ncf_type(cls, **pattern):
|
||||
pool = Pool()
|
||||
Configuration = pool.get('account.configuration')
|
||||
config = Configuration(1)
|
||||
return config.get_multivalue(
|
||||
'default_supplier_ncf_type', **pattern)
|
||||
|
||||
|
||||
|
||||
|
||||
class PartyNcfType(ModelSQL, ValueMixin):
|
||||
"Party Ncf Type"
|
||||
__name__ = 'party.party.ncf_type'
|
||||
|
||||
party = fields.Many2One(
|
||||
'party.party', "Party", ondelete='CASCADE')
|
||||
customer_ncf_type = customer_ncf_type
|
||||
supplier_ncf_type = supplier_ncf_type
|
||||
|
||||
@classmethod
|
||||
def __register__(cls, module_name):
|
||||
"""
|
||||
Migrate legacy party NCF fields safely.
|
||||
|
||||
Old structure:
|
||||
party_party_ncf_type.customer_ncf_type = ncf.manager.sequence.id
|
||||
party_party_ncf_type.supplier_ncf_type = ncf.manager.sequence.id
|
||||
|
||||
New structure:
|
||||
party_party_ncf_type.customer_ncf_type = NCF type code
|
||||
party_party_ncf_type.supplier_ncf_type = NCF type code
|
||||
|
||||
The old integer columns must be renamed before super().__register__(),
|
||||
otherwise Tryton tries to convert int4 -> varchar directly.
|
||||
"""
|
||||
cls._ncf_prepare_legacy_party_columns()
|
||||
super().__register__(module_name)
|
||||
cls._ncf_migrate_party_ncf_types_from_legacy_columns()
|
||||
|
||||
@classmethod
|
||||
def _ncf_column_exists(cls, cursor, table_name, column_name):
|
||||
if _backend_name() == 'sqlite':
|
||||
cursor.execute(f'PRAGMA table_info("{table_name}")')
|
||||
return any(row[1] == column_name for row in cursor.fetchall())
|
||||
cursor.execute("""
|
||||
SELECT 1
|
||||
FROM information_schema.columns
|
||||
WHERE table_name = %s
|
||||
AND column_name = %s
|
||||
""", (table_name, column_name))
|
||||
return cursor.fetchone() is not None
|
||||
|
||||
@classmethod
|
||||
def _ncf_column_udt_name(cls, cursor, table_name, column_name):
|
||||
if _backend_name() == 'sqlite':
|
||||
cursor.execute(f'PRAGMA table_info("{table_name}")')
|
||||
for row in cursor.fetchall():
|
||||
if row[1] == column_name:
|
||||
return (row[2] or '').lower()
|
||||
return None
|
||||
cursor.execute("""
|
||||
SELECT udt_name
|
||||
FROM information_schema.columns
|
||||
WHERE table_name = %s
|
||||
AND column_name = %s
|
||||
""", (table_name, column_name))
|
||||
row = cursor.fetchone()
|
||||
return row[0] if row else None
|
||||
|
||||
@classmethod
|
||||
def _ncf_prepare_legacy_party_columns(cls):
|
||||
"""
|
||||
Rename old integer columns before Tryton creates the new Selection
|
||||
columns.
|
||||
"""
|
||||
cursor = Transaction().connection.cursor()
|
||||
|
||||
table_name = 'party_party_ncf_type'
|
||||
|
||||
columns = (
|
||||
('customer_ncf_type', 'customer_ncf_type_legacy_sequence_id'),
|
||||
('supplier_ncf_type', 'supplier_ncf_type_legacy_sequence_id'),
|
||||
)
|
||||
|
||||
for old_column, legacy_column in columns:
|
||||
if not cls._ncf_column_exists(cursor, table_name, old_column):
|
||||
continue
|
||||
|
||||
if cls._ncf_column_exists(cursor, table_name, legacy_column):
|
||||
continue
|
||||
|
||||
udt_name = cls._ncf_column_udt_name(
|
||||
cursor, table_name, old_column)
|
||||
|
||||
if udt_name in {'int2', 'int4', 'int8', 'integer'}:
|
||||
logger.info(
|
||||
"Renaming legacy %s.%s integer column to %s before "
|
||||
"Tryton creates the new Selection column",
|
||||
table_name,
|
||||
old_column,
|
||||
legacy_column,
|
||||
)
|
||||
|
||||
cursor.execute(f"""
|
||||
ALTER TABLE {table_name}
|
||||
RENAME COLUMN {old_column} TO {legacy_column}
|
||||
""")
|
||||
|
||||
@classmethod
|
||||
def _ncf_migrate_party_ncf_types_from_legacy_columns(cls):
|
||||
"""
|
||||
Convert legacy party NCF sequence IDs into real NCF type values.
|
||||
|
||||
Example:
|
||||
customer_ncf_type_legacy_sequence_id = 1
|
||||
ncf_manager_sequence.id = 1
|
||||
ncf_manager_sequence.ncf_type = '31'
|
||||
|
||||
Result:
|
||||
customer_ncf_type = '31'
|
||||
"""
|
||||
cursor = Transaction().connection.cursor()
|
||||
|
||||
table_name = 'party_party_ncf_type'
|
||||
|
||||
columns = (
|
||||
('customer_ncf_type', 'customer_ncf_type_legacy_sequence_id'),
|
||||
('supplier_ncf_type', 'supplier_ncf_type_legacy_sequence_id'),
|
||||
)
|
||||
|
||||
for new_column, legacy_column in columns:
|
||||
if not cls._ncf_column_exists(cursor, table_name, legacy_column):
|
||||
logger.info(
|
||||
"No legacy column %s found in %s. Skipping migration.",
|
||||
legacy_column,
|
||||
table_name,
|
||||
)
|
||||
continue
|
||||
|
||||
logger.info(
|
||||
"Migrating party field %s from legacy sequence ID to "
|
||||
"fiscal NCF type code",
|
||||
new_column,
|
||||
)
|
||||
|
||||
cursor.execute(f"""
|
||||
UPDATE {table_name} AS party_ncf
|
||||
SET {new_column} = seq.ncf_type
|
||||
FROM ncf_manager_sequence AS seq
|
||||
WHERE party_ncf.{legacy_column} IS NOT NULL
|
||||
AND party_ncf.{legacy_column} = seq.id
|
||||
AND seq.ncf_type IS NOT NULL
|
||||
AND seq.ncf_type <> ''
|
||||
AND (
|
||||
party_ncf.{new_column} IS NULL
|
||||
OR party_ncf.{new_column} = ''
|
||||
OR party_ncf.{new_column} <> seq.ncf_type
|
||||
)
|
||||
""")
|
||||
|
||||
logger.info(
|
||||
"Migrated %s rows for party field %s",
|
||||
cursor.rowcount,
|
||||
new_column,
|
||||
)
|
||||
|
||||
cursor.execute(f"""
|
||||
SELECT COUNT(*)
|
||||
FROM {table_name} AS party_ncf
|
||||
LEFT JOIN ncf_manager_sequence AS seq
|
||||
ON party_ncf.{legacy_column} = seq.id
|
||||
WHERE party_ncf.{legacy_column} IS NOT NULL
|
||||
AND seq.id IS NULL
|
||||
""")
|
||||
|
||||
unmatched = cursor.fetchone()[0]
|
||||
|
||||
if unmatched:
|
||||
logger.warning(
|
||||
"%s rows in %s have legacy values in %s that do not "
|
||||
"match any ncf_manager_sequence.id",
|
||||
unmatched,
|
||||
table_name,
|
||||
legacy_column,
|
||||
)
|
||||
Reference in New Issue
Block a user