Odoo Migration Guide: From v14 to v16 Best Practices

# migration_analysis.py
import os
import ast
import re

class MigrationAnalyzer:
    def __init__(self, odoo14_path, odoo16_path):
        self.odoo14_path = odoo14_path
        self.odoo16_path = odoo16_path
        self.issues = []
    
    def analyze_custom_modules(self):
        """Analyze custom modules for migration issues"""
        custom_modules = []
        
        for root, dirs, files in os.walk(self.odoo14_path):
            if '__manifest__.py' in files:
                module_path = os.path.relpath(root, self.odoo14_path)
                if not any(module_path.startswith(standard) for standard in ['odoo/addons', 'addons']):
                    custom_modules.append(module_path)
        
        for module in custom_modules:
            self.analyze_module(module)
        
        return self.issues
    
    def analyze_module(self, module_path):
        """Analyze individual module for migration issues"""
        full_path = os.path.join(self.odoo14_path, module_path)
        
        # Check manifest
        manifest_path = os.path.join(full_path, '__manifest__.py')
        if os.path.exists(manifest_path):
            self.check_manifest(manifest_path)
        
        # Analyze Python files
        for root, dirs, files in os.walk(full_path):
            for file in files:
                if file.endswith('.py'):
                    self.analyze_python_file(os.path.join(root, file))
                elif file.endswith('.xml'):
                    self.analyze_xml_file(os.path.join(root, file))
    
    def check_manifest(self, manifest_path):
        """Check manifest file for compatibility issues"""
        with open(manifest_path, 'r') as f:
            manifest = ast.literal_eval(f.read())
        
        # Check dependencies
        if 'depends' in manifest:
            for dep in manifest['depends']:
                if self.is_removed_module(dep):
                    self.issues.append({
                        'type': 'dependency',
                        'file': manifest_path,
                        'issue': f'Removed dependency: {dep}',
                        'severity': 'high'
                    })
        
        # Check external dependencies
        if 'external_dependencies' in manifest:
            self.check_external_dependencies(manifest['external_dependencies'])
    
    def analyze_python_file(self, file_path):
        """Analyze Python file for syntax changes"""
        with open(file_path, 'r', encoding='utf-8') as f:
            content = f.read()
        
        # Check for deprecated methods
        deprecated_patterns = [
            (r'\.browse\s*\(\s*.*\s*,\s*.*\s*\)', 'browse() with context parameter is deprecated'),
            (r'\.search\s*\(\s*.*\s*,\s*.*\s*,\s*.*\s*\)', 'search() with offset/limit syntax changed'),
            (r'cr\.execute\s*\(\s*.*%s.*\)', 'String formatting in SQL queries - use params instead'),
            (r'_defaults\s*=', '_defaults is deprecated, use default_get instead'),
            (r'columns\s*=', 'columns definition is deprecated, use _rec_name/_order fields'),
        ]
        
        for pattern, message in deprecated_patterns:
            if re.search(pattern, content):
                self.issues.append({
                    'type': 'deprecated',
                    'file': file_path,
                    'issue': message,
                    'severity': 'medium'
                })
    
    def is_removed_module(self, module):
        """Check if module was removed in v16"""
        removed_modules = [
            'website_theme_install',
            'account_bank_statement_import_camt',
            # Add more removed modules as needed
        ]
        return module in removed_modules

# Usage
analyzer = MigrationAnalyzer('/path/to/odoo14', '/path/to/odoo16')
issues = analyzer.analyze_custom_modules()
for issue in issues:
    print(f"[{issue['severity'].upper()}] {issue['file']}: {issue['issue']}")

# migration_script.py
import psycopg2
import logging
from contextlib import contextmanager

class DatabaseMigrator:
    def __init__(self, db_config):
        self.db_config = db_config
        self.logger = logging.getLogger(__name__)
    
    @contextmanager
    def db_connection(self):
        """Database connection context manager"""
        conn = psycopg2.connect(**self.db_config)
        conn.autocommit = False
        try:
            yield conn
        except Exception:
            conn.rollback()
            raise
        finally:
            conn.close()
    
    def create_backup(self, db_name):
        """Create database backup before migration"""
        import subprocess
        import datetime
        
        timestamp = datetime.datetime.now().strftime('%Y%m%d_%H%M%S')
        backup_file = f"{db_name}_pre_migration_{timestamp}.sql"
        
        cmd = [
            'pg_dump',
            '-h', self.db_config['host'],
            '-p', str(self.db_config['port']),
            '-U', self.db_config['user'],
            '-d', db_name,
            '-f', backup_file
        ]
        
        try:
            subprocess.run(cmd, check=True)
            self.logger.info(f"Backup created: {backup_file}")
            return backup_file
        except subprocess.CalledProcessError as e:
            self.logger.error(f"Backup failed: {e}")
            raise
    
    def migrate_database(self, source_db, target_db):
        """Execute database migration"""
        with self.db_connection() as conn:
            cursor = conn.cursor()
            
            try:
                # 1. Update ir_module_module version
                self.update_module_versions(cursor)
                
                # 2. Handle model changes
                self.migrate_models(cursor)
                
                # 3. Update field definitions
                self.migrate_fields(cursor)
                
                # 4. Handle data migrations
                self.migrate_data(cursor)
                
                # 5. Update views and menus
                self.migrate_views(cursor)
                
                conn.commit()
                self.logger.info("Database migration completed successfully")
                
            except Exception as e:
                conn.rollback()
                self.logger.error(f"Migration failed: {e}")
                raise
    
    def update_module_versions(self, cursor):
        """Update module versions for v16"""
        updates = [
            ("base", "16.0.1.0"),
            ("web", "16.0.1.0"),
            # Add other standard modules
        ]
        
        for module, version in updates:
            cursor.execute("""
                UPDATE ir_module_module 
                SET latest_version = %s, installed_version = %s
                WHERE name = %s
            """, (version, version, module))
    
    def migrate_models(self, cursor):
        """Handle model migrations"""
        # Example: Handle removed models
        removed_models = [
            'mail.bot',
            # Add other removed models
        ]
        
        for model in removed_models:
            cursor.execute("""
                DELETE FROM ir_model 
                WHERE model = %s
            """, (model,))
            
            cursor.execute("""
                DELETE FROM ir_model_fields 
                WHERE model = %s
            """, (model,))
    
    def migrate_fields(self, cursor):
        """Handle field migrations"""
        # Example: Handle field type changes
        field_changes = [
            ('res.partner', 'ref', 'char', 'New reference field format'),
            # Add other field changes
        ]
        
        for model, field_name, field_type, description in field_changes:
            cursor.execute("""
                UPDATE ir_model_fields 
                SET ttype = %s, field_description = %s
                WHERE model = %s AND name = %s
            """, (field_type, description, model, field_name))
    
    def migrate_data(self, cursor):
        """Handle data migrations"""
        # Example: Update user preferences format
        cursor.execute("""
            UPDATE res_users 
            SET barcode = NULL 
            WHERE barcode IS NOT NULL AND barcode = ''
        """)
        
        # Example: Migrate configuration settings
        cursor.execute("""
            UPDATE ir_config_parameter 
            SET value = 'web' 
            WHERE key = 'web.base.url' AND value LIKE 'http%'
        """)
    
    def migrate_views(self, cursor):
        """Handle view migrations"""
        # Update deprecated view attributes
        cursor.execute("""
            UPDATE ir_ui_view 
            SET arch_db = REPLACE(arch_db, 'colors="', 'decoration-')
            WHERE arch_db LIKE '%colors="%'
        """)
        
        # Update button attributes
        cursor.execute("""
            UPDATE ir_ui_view 
            SET arch_db = REPLACE(arch_db, 'string="', 'name="')
            WHERE arch_db LIKE '%button%' AND arch_db LIKE '%string="%'
        """)

# Code migration examples

# 1. ORM method changes
# Old (v14):
records = self.env['model'].browse(ids, context={'lang': 'en_US'})

# New (v16):
records = self.env['model'].with_context(lang='en_US').browse(ids)

# 2. Search method changes
# Old (v14):
records = self.env['model'].search(domain, offset=10, limit=20)

# New (v16):
records = self.env['model'].search(domain, offset=10, limit=20)
# Note: Syntax remains same but performance improved

# 3. Field definition changes
# Old (v14):
class MyModel(models.Model):
    _name = 'my.model'
    
    name = fields.Char('Name')
    description = fields.Text('Description')
    
    _defaults = {
        'name': 'Default Name',
    }

# New (v16):
class MyModel(models.Model):
    _name = 'my.model'
    
    name = fields.Char('Name', default='Default Name')
    description = fields.Text('Description')

# 4. API method changes
# Old (v14):
@api.model
def create(self, vals):
    if self._context.get('default_name'):
        vals['name'] = self._context['default_name']
    return super(MyModel, self).create(vals)

# New (v16):
@api.model
def create(self, vals):
    if self.env.context.get('default_name'):
        vals['name'] = self.env.context['default_name']
    return super().create(vals)

# tests/test_migration.py
from odoo.tests.common import TransactionCase
from odoo.exceptions import ValidationError

class MigrationTest(TransactionCase):
    
    def setUp(self):
        super().setUp()
        # Setup test data
        self.partner = self.env['res.partner'].create({
            'name': 'Test Partner',
            'email': 'test@example.com',
        })
    
    def test_model_functionality(self):
        """Test that models work correctly after migration"""
        # Test basic CRUD operations
        partner_count = self.env['res.partner'].search_count([])
        self.assertGreater(partner_count, 0)
        
        # Test field access
        self.assertEqual(self.partner.name, 'Test Partner')
        self.assertEqual(self.partner.email, 'test@example.com')
    
    def test_workflow_functionality(self):
        """Test that workflows still work"""
        # Test sale order workflow
        sale_order = self.env['sale.order'].create({
            'partner_id': self.partner.id,
            'order_line': [(0, 0, {
                'product_id': self.env.ref('product.product_product_1').id,
                'product_uom_qty': 1,
                'price_unit': 100.0,
            })]
        })
        
        # Test state changes
        sale_order.action_confirm()
        self.assertEqual(sale_order.state, 'sale')
    
    def test_api_compatibility(self):
        """Test API endpoints work correctly"""
        # Test JSON-RPC calls
        result = self.env['res.partner'].search_read([
            ('id', '=', self.partner.id)
        ], ['name', 'email'])
        
        self.assertEqual(len(result), 1)
        self.assertEqual(result[0]['name'], 'Test Partner')
    
    def test_permissions(self):
        """Test that permissions are preserved"""
        # Test access rights
        user = self.env['res.users'].create({
            'name': 'Test User',
            'login': 'testuser',
            'password': 'testpass',
        })
        
        # Test with different user contexts
        as_user = self.env(user=user)
        with self.assertRaises(AccessError):
            as_user.env['res.users'].search([])

# performance_monitor.py
import time
import logging
from functools import wraps

class PerformanceMonitor:
    def __init__(self):
        self.logger = logging.getLogger(__name__)
    
    def monitor_method(self, method_name):
        """Decorator to monitor method performance"""
        def decorator(func):
            @wraps(func)
            def wrapper(*args, **kwargs):
                start_time = time.time()
                try:
                    result = func(*args, **kwargs)
                    return result
                finally:
                    end_time = time.time()
                    duration = end_time - start_time
                    
                    if duration > 1.0:  # Log slow methods
                        self.logger.warning(
                            f"Slow method detected: {method_name} took {duration:.2f}s"
                        )
                    
                    # Store performance metrics
                    self.store_metrics(method_name, duration)
            return wrapper
        return decorator
    
    def store_metrics(self, method_name, duration):
        """Store performance metrics for analysis"""
        # Implement metrics storage (Redis, database, etc.)
        pass

# Usage in models
monitor = PerformanceMonitor()

class ResPartner(models.Model):
    _name = 'res.partner'
    
    @monitor.monitor_method('partner_search')
    @api.model
    def search(self, args, offset=0, limit=None, order=None, count=False):
        return super().search(args, offset, limit, order, count)

# rollback_plan.py
class RollbackManager:
    def __init__(self, backup_config):
        self.backup_config = backup_config
    
    def create_rollback_point(self, db_name):
        """Create a rollback point before major changes"""
        import subprocess
        
        timestamp = time.strftime('%Y%m%d_%H%M%S')
        rollback_file = f"{db_name}_rollback_{timestamp}.sql"
        
        # Create incremental backup
        cmd = [
            'pg_dump',
            '--format=custom',
            '--verbose',
            '--file', rollback_file,
            '--host', self.backup_config['host'],
            '--port', str(self.backup_config['port']),
            '--username', self.backup_config['user'],
            db_name
        ]
        
        subprocess.run(cmd, check=True)
        return rollback_file
    
    def execute_rollback(self, rollback_file, target_db):
        """Execute rollback from backup file"""
        import subprocess
        
        cmd = [
            'pg_restore',
            '--clean',
            '--if-exists',
            '--verbose',
            '--host', self.backup_config['host'],
            '--port', str(self.backup_config['port']),
            '--username', self.backup_config['user'],
            '--dbname', target_db,
            rollback_file
        ]
        
        subprocess.run(cmd, check=True)