Connecting to a supplier's database: direct import without files

Benefits of Direct Database Connection

Database Direct Connection is an integration method where your system reads data directly from the vendor's DBMS without using intermediate files.

Key benefits:

• Real-time relevance — data is always up-to-date, without delays in file generation and transfer
• No problems with formats — no need to parse XLS, CSV, or XML, the data is already structured
• Sample control — you decide which fields and records to load via SQL queries
• Incremental updates - easy to download only changed data by timestamp
• Saving resources - no need to store and process files
• Automation — requests are executed on schedule without human intervention
• Performance — SQL queries often work faster than parsing large files

Comparison with file sharing:

When to use direct database access

A direct database connection is a powerful tool, but it's not always practical. Let's look at scenarios where this solution is truly necessary.

Ideal scenarios for direct connection:

1. High update frequency

If prices or inventory changes every 15-30 minutes, file sharing becomes a bottleneck. A direct connection allows for data requests to be made minute-by-minute or even on demand.

• Financial instruments with dynamic pricing
• Fast-moving goods (electronics, equipment)
• Products with limited stock

2. Big data

A catalog of 100,000+ SKUs in a file takes a significant amount of time to generate, transfer, and parse. A SQL query to the database retrieves only the required data in seconds.

3. Complex data structure

When products have multiple relationships (categories, attributes, modifications, warehouses), a relational database is the natural storage format. Files require denormalization and duplication.

4. Long-term partnership

Direct connection requires a trusting relationship and technical support from the supplier. This is justified for strategic partners, not one-time deals.

5. The need for bilateral exchange

When you need to not only receive data but also transmit it back (order statuses, product reservations), an API or direct connection to the database is more efficient than file sharing.

When NOT to use a direct connection:

• The supplier is not ready to provide access — many companies do not allow external connections to their databases for security reasons
• One-time or infrequent downloads - if the update is needed once a day or less often, the file is simpler
• Lack of technical competence — setup requires knowledge of SQL, network security, and database administration
• Unstable supplier infrastructure - If its base is often unavailable, this will create problems
• There is no way to ensure security — connecting without a VPN or SSH tunnel creates risks

Important: Connecting directly to a database is a technical and legal matter. Be sure to enter into an agreement with the provider that defines access rights, responsibilities, and security measures.

Security requirements when connecting to a database

Security is a critical aspect when connecting directly to a database. Incorrect configuration can lead to data leakage, unauthorized access, or damage to information.

Basic safety principles:

1. Read-Only Access

Create a database user with read-only permissions. This prevents accidental or deliberate modification of supplier information.

 CREATE USER 'readonly_user'@'%' IDENTIFIED BY 'strong_password_here'; GRANT SELECT ON supplier_db.products TO 'readonly_user'@'%'; GRANT SELECT ON supplier_db.prices TO 'readonly_user'@'%'; GRANT SELECT ON supplier_db.stock TO 'readonly_user'@'%'; FLUSH PRIVILEGES;

 CREATE USER readonly_user WITH PASSWORD 'strong_password_here'; GRANT CONNECT ON DATABASE supplier_db TO readonly_user; GRANT USAGE ON SCHEMA public TO readonly_user; GRANT SELECT ON products, prices, stock TO readonly_user;

2. IP Whitelisting

Restrict database access to specific IP addresses on your system. This prevents connections from unauthorized sources.

 bind-address = 0.0.0.0 # Then use firewall to restrict IP

 sudo ufw allow from 203.0.113.10 to any port 3306 sudo ufw allow from 198.51.100.25 to any port 3306

3. Using a VPN or SSH tunnel

The most secure method is to connect via a secure channel. This encrypts all data and hides the real database address.

 ssh -L 3307:localhost:3306 user@supplier-server.com # Now connect to localhost:3307

4. SSL/TLS connection encryption

If a direct connection over the Internet is unavoidable, use SSL to encrypt traffic.

 mysql://readonly_user:password@supplier.com:3306/database?ssl-mode=REQUIRED

5. Strong passwords and rotation

• Minimum 16 characters, including letters, numbers, and special characters
• Change your password every 3-6 months
• Don't use the same passwords for different providers
• Store passwords encrypted (not in code!)

6. Monitoring and logging

Monitor all database connections and queries:

• Log connection time, IP address, and completed requests
• Set up alerts for abnormal activity
• Regularly check your logs for suspicious activity.

Pre-launch safety checklist:

1. A user with only SELECT rights has been created.
2. IP whitelist configured
3. A VPN, SSH tunnel, or SSL is used
4. Password generated automatically (minimum 16 characters)
5. The connection was tested in an isolated environment.
6. Logging of all operations is configured
7. An access agreement has been signed with the supplier
8. An action plan has been defined in case of access compromise.

Critical: Never store database passwords in plaintext in code or configuration files. Use environment variables, Kubernetes secrets, AWS Secrets Manager, or similar solutions.

Secure connection methods

Let's look at three main methods for establishing a secure connection to a provider's database.

Method 1: Direct Connection with IP Whitelist and SSL

The simplest method, suitable for the initial level of security.

Operating principle:

1. The provider provides access to the database via the Internet (port 3306 for MySQL, 5432 for PostgreSQL)
2. Your firewall has a whitelist of your IP addresses.
3. The connection uses SSL/TLS encryption.
4. Password authentication

Pros:

• Easy setup
• Does not require additional infrastructure
• Suitable for small projects

Cons:

• The database is open to the Internet (even with restrictions)
• Dependence on static IPs (problems when changing hosting)
• Limited protection

Method 2: VPN tunnel

Create a secure virtual network between your server and the provider's server.

Operating principle:

1. Setting up a VPN connection (OpenVPN, WireGuard, IPSec)
2. Your server and the provider's server are in the same virtual network
3. The database is only accessible within the VPN.
4. All traffic is encrypted at the VPN level.

OpenVPN setup example (basic configuration):

 client dev tun proto udp remote supplier-vpn.com 1194 resolv-retry infinite nobind persist-key persist-tun ca ca.crt cert client.crt key client.key cipher AES-256-CBC verb 3

Pros:

• High level of security
• The database is not open to the Internet.
• Encryption of all traffic
• Suitable for corporate integration

Cons:

• Complex setup
• Requires technical knowledge on both sides
• Additional point of failure

Method 3: SSH tunnel

The most flexible method uses SSH to forward the database port.

Operating principle:

1. You connect via SSH to the provider's server
2. SSH forwards the database port to your local server.
3. The application connects to localhost, but actually works with a remote database.
4. All traffic goes through an encrypted SSH tunnel.

Command to create SSH tunnel:

 # For MySQL ssh -f -N -L 3307:127.0.0.1:3306 user@supplier-server.com # For PostgreSQL ssh -f -N -L 5433:127.0.0.1:5432 user@supplier-server.com # For MSSQL ssh -f -N -L 1434:127.0.0.1:1433 user@supplier-server.com

Command parameters:

• -f - run in the background
• -N - do not execute commands (tunnel only)
• -L — local port: remote host: remote port

Usage in code (Python example):

 import pymysql from sshtunnel import SSHTunnelForwarder with SSHTunnelForwarder( ('supplier-server.com', 22), ssh_username='ssh_user', ssh_private_key='/path/to/private_key', remote_bind_address=('127.0.0.1', 3306) ) as tunnel: connection = pymysql.connect( host='127.0.0.1', port=tunnel.local_bind_port, user='readonly_user', password='password', database='supplier_db' ) # Working with the database

Pros:

• Very high security
• Flexibility in customization
• SSH access is easier to obtain than VPN
• Does not require changes to the database configuration

Cons:

• Dependence on the stability of the SSH connection
• SSH access to the provider's server is required
• The need for SSH key management

Recommendations for choosing a method:

Step-by-step connection setup

Let's look at the full process of setting up a direct connection to a supplier's database, from gaining access to the first successful import.

Step 1: Coordination with the supplier

Before technical implementation, be sure to agree on all details with the supplier:

What you need to know:

• DBMS type: MySQL, PostgreSQL, MSSQL, Oracle
• Version: important for driver compatibility
• Host and port: database server address
• Connection method: direct, VPN, SSH
• Data structure: table names, schemas, key fields
• Access rights: which tables/views are accessed
• Restrictions: request frequency, data volume limits
• Contact person: Who can help with technical problems?

What needs to be agreed upon:

• Creating a read-only user
• Your IP addresses for whitelist
• Database maintenance schedule (when downtime is possible)
• Procedure for notification of changes in structure

Step 2: Obtaining and Safely Storing Access

Once approved, you will receive credentials:

 Host: db.supplier.com Port: 3306 Database: supplier_products Username: partner_readonly Password: xK9#mP2$vL8@nQ5!

Safe storage:

Using environment variables (.env file):

 DB_SUPPLIER_HOST=db.supplier.com DB_SUPPLIER_PORT=3306 DB_SUPPLIER_NAME=supplier_products DB_SUPPLIER_USER=partner_readonly DB_SUPPLIER_PASS=xK9#mP2$vL8@nQ5!

In the code (Python example):

 import os from dotenv import load_dotenv load_dotenv() db_config ={ 'host': os.getenv('DB_SUPPLIER_HOST'), 'port': int(os.getenv('DB_SUPPLIER_PORT')), 'database': os.getenv('DB_SUPPLIER_NAME'), 'user': os.getenv('DB_SUPPLIER_USER'), 'password': os.getenv('DB_SUPPLIER_PASS') }

Step 3: Installing the required drivers

To work with various DBMS, you need the appropriate libraries:

MySQL:

 # Python pip install mysql-connector-python pymysql # Node.js npm install mysql2 # PHP (usually already installed) apt-get install php-mysql

PostgreSQL:

 # Python pip install psycopg2-binary # Node.js npm install pg # PHP apt-get install php-pgsql

Microsoft SQL Server:

 # Python pip install pymssql pyodbc # Node.js npm install mssql # PHP apt-get install php-sqlsrv

Step 4: Test connection

Before writing import logic, make sure the connection works.

MySQL connection test (Python):

 import mysql.connector from mysql.connector import Error try: connection = mysql.connector.connect( host='db.supplier.com', port=3306, database='supplier_products', user='partner_readonly', password='xK9#mP2$vL8@nQ5!', ssl_disabled=False # Use SSL if available ) if connection.is_connected(): db_info = connection.get_server_info() print(f"Successfully connected to MySQL Server version {db_info}") cursor = connection.cursor() cursor.execute("SELECT DATABASE();") record = cursor.fetchone() print(f"You're connected to database: {record}") except Error as e: print(f"Error while connecting to MySQL: {e}") finally: if connection.is_connected(): cursor.close() connection.close() print("MySQL connection is closed")

PostgreSQL connection test (Python):

 import psycopg2 try: connection = psycopg2.connect( host='db.supplier.com', port=5432, database='supplier_products', user='partner_readonly', password='xK9#mP2$vL8@nQ5!', sslmode='require' # Require SSL ) cursor = connection.cursor() cursor.execute("SELECT version();") record = cursor.fetchone() print(f"You are connected to - {record}") except Exception as error: print(f"Error while connecting to PostgreSQL: {error}") finally: if connection: cursor.close() connection.close() print("PostgreSQL connection is closed")

Step 5: Examine the data structure

Find out what tables are available and how they are related:

Getting a list of tables (MySQL):

 SHOW TABLES;

Getting table structure (MySQL):

 DESCRIBE products; SHOW CREATE TABLE products;

Getting a list of tables (PostgreSQL):

 SELECT table_name FROM information_schema.tables WHERE table_schema = 'public';

Getting the table structure (PostgreSQL):

 SELECT column_name, data_type, character_maximum_length FROM information_schema.columns WHERE table_name = 'products';

Step 6: Writing SQL queries to retrieve data

Create queries to retrieve the data you need (examples in the next section).

Step 7: Implementing the ETL Process

Extract, Transform, Load - the standard data import process.

Step 8: Setting up automation

Set up scheduled import execution (cron, task scheduler, orchestrators like Airflow).

Example of a cron task (Linux):

 # Run every 30 minutes */30 * * * * /usr/bin/python3 /path/to/import_script.py >> /var/log/import.log 2>&1 # Run every hour at 15 minutes past 15 * * * * /usr/bin/python3 /path/to/import_script.py # Run every day at 3:00 AM 0 3 * * * /usr/bin/python3 /path/to/import_script.py

Advice: Start with infrequent runs (every few hours), ensure the process is stable, and then increase the frequency. Avoid placing excessive load on the provider's database.

Examples of SQL queries for data retrieval

Efficient SQL queries are the key to fast and accurate data import. Let's look at typical scenarios and optimized queries.

Basic sample of all products

 SELECT p.id AS product_id, p.sku AS article, p.name AS product_name, p.description, p.brand, c.name AS category_name, pr.price_retail AS price, pr.price_wholesale AS wholesale_price, s.quantity AS stock_quantity, p.created_at, p.updated_at FROM products p LEFT JOIN categories c ON p.category_id = c.id LEFT JOIN prices pr ON p.id = pr.product_id LEFT JOIN stock s ON p.id = s.product_id WHERE p.active = 1 ORDER BY p.id;

Incremental sampling (changed products only)

The most efficient way is to load only the data that has been updated since the last import:

 SELECT p.id, p.sku, p.name, p.description, pr.price_retail, pr.price_wholesale, s.quantity, p.updated_at FROM products p LEFT JOIN prices pr ON p.id = pr.product_id LEFT JOIN stock s ON p.id = s.product_id WHERE p.updated_at > '2025-10-20 12:00:00' OR pr.updated_at > '2025-10-20 12:00:00' OR s.updated_at > '2025-10-20 12:00:00' ORDER BY p.updated_at DESC;

How to use: Store the time of the last successful import and use it in the next request.

Selection filtered by category

 SELECT p.id, p.sku, p.name, pr.price_retail, s.quantity FROM products p INNER JOIN categories c ON p.category_id = c.id LEFT JOIN prices pr ON p.id = pr.product_id LEFT JOIN stock s ON p.id = s.product_id WHERE c.name IN ('Electronics', 'Computers', 'Smartphones') AND p.active = 1 AND s.quantity > 0;

Sample with aggregation (total balances by warehouse)

 SELECT p.id, p.sku, p.name, pr.price_retail, SUM(s.quantity) AS total_stock, COUNT(DISTINCT s.warehouse_id) AS warehouses_count FROM products p LEFT JOIN prices pr ON p.id = pr.product_id LEFT JOIN stock s ON p.id = s.product_id WHERE p.active = 1 GROUP BY p.id, p.sku, p.name, pr.price_retail HAVING SUM(s.quantity) > 0;

Currency-based sampling

 SELECT p.id, p.sku, p.name, pr.price AS original_price, pr.currency AS original_currency, CASE WHEN pr.currency = 'USD' THEN pr.price WHEN pr.currency = 'EUR' THEN pr.price * 1.08 WHEN pr.currency = 'UAH' THEN pr.price / 41.5 ELSE pr.price END AS price_usd FROM products p LEFT JOIN prices pr ON p.id = pr.product_id WHERE p.active = 1;

Pagination for large tables

When working with millions of records, load data in chunks:

MySQL / PostgreSQL:

 SELECT p.id, p.sku, p.name, pr.price_retail, s.quantity FROM products p LEFT JOIN prices pr ON p.id = pr.product_id LEFT JOIN stock s ON p.id = s.product_id WHERE p.active = 1 ORDER BY p.id LIMIT 1000 OFFSET 0; -- First page -- Next pages: OFFSET 1000, OFFSET 2000, etc.

MSSQL:

 SELECT p.id, p.sku, p.name, pr.price_retail, s.quantity FROM products p LEFT JOIN prices pr ON p.id = pr.product_id LEFT JOIN stock s ON p.id = s.product_id WHERE p.active = 1 ORDER BY p.id OFFSET 0 ROWS FETCH NEXT 1000 ROWS ONLY;

Query optimization

Use indexes:

Make sure the fields in WHERE, JOIN, and ORDER BY are indexed:

 -- Checking indexes (MySQL) SHOW INDEX FROM products; -- Checking indexes (PostgreSQL) SELECT indexname, indexdef FROM pg_indexes WHERE tablename = 'products';

Select only the fields you need:

Avoid SELECT *, specify specific columns.

Use EXPLAIN to analyze:

 EXPLAIN SELECT p.id, p.sku, p.name FROM products p WHERE p.updated_at > '2025-10-20';

Handling large results in code

Cursors for batch processing (Python):

 import mysql.connector connection = mysql.connector.connect(**db_config) cursor = connection.cursor(dictionary=True) query = """ SELECT p.id, p.sku, p.name, pr.price_retail, s.quantity FROM products p LEFT JOIN prices pr ON p.id = pr.product_id LEFT JOIN stock s ON p.id = s.product_id WHERE p.active = 1 """ cursor.execute(query) batch_size = 1000 while True: rows = cursor.fetchmany(batch_size) if not rows: break # Processing a batch of data for row in rows: process_product(row) print(f"Processed {len(rows)} products") cursor.close() connection.close()

Important: Agree with your provider on the optimal sample size and query frequency. Excessively frequent or heavy queries may put a strain on their infrastructure.

Working with different types of DBMS

MySQL, PostgreSQL, and Microsoft SQL Server each have their own connection and operating requirements. Let's look at the specifics of each system.

MySQL / MariaDB

Peculiarities:

• The most widely used DBMS for web applications
• Easy to set up and use
• Good reading performance
• Standard port: 3306

Connection string:

 mysql://username:password@host:3306/database?charset=utf8mb4&ssl=true

Connection example (Python):

 import mysql.connector config ={ 'host': 'db.supplier.com', 'port': 3306, 'database': 'supplier_db', 'user': 'readonly_user', 'password': 'password', 'charset': 'utf8mb4', 'use_unicode': True, 'ssl_disabled': False } connection = mysql.connector.connect(**config) cursor = connection.cursor(dictionary=True) cursor.execute("SELECT * FROM products LIMIT 10") results = cursor.fetchall() for row in results: print(row) cursor.close() connection.close()

SQL Features:

• The register of the tables depends on the OS (Windows - not important, Linux - important)
• Usage LIMIT for pagination
• AUTO_INCREMENT for auto-increment
• Backing agents for shielding: `table_name`

PostgreSQL

Peculiarities:

• Stronger compliance with SQL standards
• Better support for complex data types (JSON, arrays)
• Excellent performance for complex queries
• Standard port: 5432

Connection string:

 postgresql://username:password@host:5432/database?sslmode=require

Connection example (Python):

 import psycopg2 from psycopg2.extras import RealDictCursor connection = psycopg2.connect( host='db.supplier.com', port=5432, database='supplier_db', user='readonly_user', password='password', sslmode='require' ) cursor = connection.cursor(cursor_factory=RealDictCursor) cursor.execute("SELECT * FROM products LIMIT 10") results = cursor.fetchall() for row in results: print(dict(row)) cursor.close() connection.close()

SQL Features:

• Case is always important (default is lowercase)
• Usage LIMIT And OFFSET
• SERIAL or GENERATED ALWAYS AS IDENTITY for auto-increment
• Double quotes for escaping: "table_name"
• Rich JSON capabilities: data->>'field'

Working with JSON fields:

 SELECT id, sku, name, attributes->>'color' AS color, attributes->>'size' AS size FROM products WHERE attributes->>'brand' = 'Samsung';

Microsoft SQL Server (MSSQL)

Peculiarities:

• Popular in the corporate environment
• Powerful tools for business analytics
• Integration with the Microsoft ecosystem
• Standard port: 1433

Connection string:

 mssql://username:password@host:1433/database?encrypt=true&trustServerCertificate=false

Connection example (Python):

 import pymssql connection = pymssql.connect( server='db.supplier.com', port=1433, database='supplier_db', user='readonly_user', password='password', tds_version='7.0' ) cursor = connection.cursor(as_dict=True) cursor.execute("SELECT TOP 10 * FROM products") results = cursor.fetchall() for row in results: print(row) cursor.close() connection.close()

Особенности SQL:

• Usage TOP N instead of LIMIT
• IDENTITY для автоинкремента
• Квадратные скобки для экранирования: [table_name]
• OFFSET... FETCH для пагинации (SQL Server 2012+)

Пагинация в MSSQL:

 -- Современный способ (SQL Server 2012+) SELECT * FROM products ORDER BY id OFFSET 0 ROWS FETCH NEXT 100 ROWS ONLY; -- Старый способ (до 2012) SELECT TOP 100 * FROM products WHERE id > @last_id ORDER BY id;

Comparison table

Универсальный подход через ORM

Для упрощения работы с разными СУБД используйте ORM (Object-Relational Mapping):

SQLAlchemy (Python) — универсальный подход:

 from sqlalchemy import create_engine, MetaData, Table from sqlalchemy.orm import sessionmaker # Работает с MySQL, PostgreSQL, MSSQL engine = create_engine('mysql://user:pass@host/db') # engine = create_engine('postgresql://user:pass@host/db') # engine = create_engine('mssql+pymssql://user:pass@host/db') metadata = MetaData() products = Table('products', metadata, autoload_with=engine) Session = sessionmaker(bind=engine) session = Session() results = session.query(products).limit(10).all() for row in results: print(dict(row))

Advantage: один код работает со всеми типами баз данных, меняется только строка подключения.

ETL процесс: Extract, Transform, Load

ETL (Extract, Transform, Load) — стандартный подход к импорту данных. Рассмотрим каждый этап применительно к импорту из базы данных поставщика.

Extract (Извлечение данных)

Первый этап — получение данных из базы данных поставщика.

Стратегии извлечения:

1. Полная выгрузка (Full Extract)

Загрузка всех данных при каждом запуске.

• Pros: простота, гарантия актуальности
• Cons: медленно для больших таблиц, высокая нагрузка
• When to use: небольшие каталоги (до 10 000 товаров)

2. Инкрементальная выгрузка (Incremental Extract)

Загрузка только измененных с последнего раза записей.

• Pros: быстро, минимальная нагрузка
• Cons: требует timestamp полей, сложнее логика
• When to use: большие каталоги, частые обновления

Пример инкрементального извлечения:

 import mysql.connector from datetime import datetime import json # Загрузить время последней синхронизации def get_last_sync_time(): try: with open('last_sync.json', 'r') as f: data = json.load(f) return data['last_sync'] except: return '2020-01-01 00:00:00' # Сохранить время текущей синхронизации def save_sync_time(sync_time): with open('last_sync.json', 'w') as f: json.dump({'last_sync': sync_time}, f) # Extract def extract_products(): connection = mysql.connector.connect(**db_config) cursor = connection.cursor(dictionary=True) last_sync = get_last_sync_time() current_sync = datetime.now().strftime('%Y-%m-%d %H:%M:%S') query = """ SELECT p.*, pr.price_retail, s.quantity FROM products p LEFT JOIN prices pr ON p.id = pr.product_id LEFT JOIN stock s ON p.id = s.product_id WHERE p.updated_at > %s OR pr.updated_at > %s OR s.updated_at > %s """ cursor.execute(query, (last_sync, last_sync, last_sync)) products = cursor.fetchall() cursor.close() connection.close() save_sync_time(current_sync) return products

Transform (Преобразование данных)

Второй этап — приведение данных к формату вашей системы.

Типовые преобразования:

1. Маппинг полей

 def transform_product(raw_product): return{ 'sku': raw_product['sku'], 'name': raw_product['product_name'].strip(), 'description': raw_product['description'], 'price': float(raw_product['price_retail']), 'stock': int(raw_product['quantity']) if raw_product['quantity'] else 0, 'category': map_category(raw_product['category_id']), 'active': raw_product['status'] == 'active' }

2. Валидация данных

 def validate_product(product): errors = [] # SKU обязателен if not product.get('sku'): errors.append('SKU is required') # Цена должна быть положительной if product.get('price', 0)<= 0: errors.append('Price must be positive') # Название не должно быть пустым if not product.get('name', '').strip(): errors.append('Name is required') return len(errors) == 0, errors

3. Обогащение данных

 def enrich_product(product): # Добавление наценки product['retail_price'] = product['price'] * 1.3 # Генерация slug для URL product['slug'] = generate_slug(product['name']) # Определение доступности product['available'] = product['stock'] > 0 # Категоризация по цене if product['price']< 100: product['price_segment'] = 'budget' elif product['price'] < 500: product['price_segment'] = 'mid-range' else: product['price_segment'] = 'premium' return product

4. Нормализация текста

 import re def normalize_text(text): if not text: return '' # Удалить лишние пробелы text = ' '.join(text.split()) # Удалить невидимые символы text = re.sub(r'[\x00-\x1F\x7F-\x9F]', '', text) # Заменить кавычки на стандартные text = text.replace('"', '"').replace('"', '"') return text.strip() def transform_product(raw): return{ 'name': normalize_text(raw['name']), 'description': normalize_text(raw['description']), #... }

5. Конвертация валют

 def convert_currency(amount, from_currency, to_currency='USD'): rates ={ 'USD': 1.0, 'EUR': 1.08, 'UAH': 0.024, 'GBP': 1.26 } # Сначала конвертируем в USD, потом в целевую валюту usd_amount = amount / rates.get(from_currency, 1.0) return usd_amount * rates.get(to_currency, 1.0) def transform_product(raw): return{ 'price_usd': convert_currency( raw['price'], raw['currency'], 'USD' ), 'price_eur': convert_currency( raw['price'], raw['currency'], 'EUR' ) }

Load (Загрузка данных)

Третий этап — сохранение преобразованных данных в вашу систему.

Стратегии загрузки:

1. Insert or Update (Upsert)

 def load_product(product, connection): cursor = connection.cursor() # MySQL query = """ INSERT INTO products (sku, name, price, stock, updated_at) VALUES (%s, %s, %s, %s, NOW()) ON DUPLICATE KEY UPDATE name = VALUES(name), price = VALUES(price), stock = VALUES(stock), updated_at = NOW() """ cursor.execute(query, ( product['sku'], product['name'], product['price'], product['stock'] )) connection.commit() cursor.close()

2. Пакетная загрузка

 def load_products_batch(products, connection): cursor = connection.cursor() query = """ INSERT INTO products (sku, name, price, stock) VALUES (%s, %s, %s, %s) ON DUPLICATE KEY UPDATE name = VALUES(name), price = VALUES(price), stock = VALUES(stock) """ # Подготовка данных для пакетной вставки data = [ (p['sku'], p['name'], p['price'], p['stock']) for p in products ] # Выполнение пакетной операции cursor.executemany(query, data) connection.commit() cursor.close() return cursor.rowcount

Полный ETL-процесс

 import logging from datetime import datetime logging.basicConfig(level=logging.INFO) logger = logging.getLogger(__name__) def run_etl_process(): start_time = datetime.now() logger.info(f"ETL process started at {start_time}") stats ={ 'extracted': 0, 'transformed': 0, 'loaded': 0, 'errors': 0 } try: # EXTRACT logger.info("Starting extraction...") raw_products = extract_products() stats['extracted'] = len(raw_products) logger.info(f"Extracted {stats['extracted']} products") # TRANSFORM logger.info("Starting transformation...") transformed_products = [] for raw_product in raw_products: try: product = transform_product(raw_product) product = enrich_product(product) is_valid, errors = validate_product(product) if is_valid: transformed_products.append(product) stats['transformed'] += 1 else: logger.warning(f"Validation failed for SKU {raw_product.get('sku')}: {errors}") stats['errors'] += 1 except Exception as e: logger.error(f"Transform error for SKU {raw_product.get('sku')}: {e}") stats['errors'] += 1 # LOAD logger.info("Starting load...") connection = get_local_db_connection() # Загружаем пакетами по 1000 записей batch_size = 1000 for i in range(0, len(transformed_products), batch_size): batch = transformed_products[i:i+batch_size] loaded = load_products_batch(batch, connection) stats['loaded'] += loaded logger.info(f"Loaded batch {i//batch_size + 1}, total loaded: {stats['loaded']}") connection.close() except Exception as e: logger.error(f"ETL process failed: {e}") raise finally: duration = (datetime.now() - start_time).total_seconds() logger.info(f"ETL process completed in{duration:.2f} seconds") logger.info(f"Stats: {stats}") # Отправка уведомления send_notification(stats, duration) return stats if __name__ == '__main__': run_etl_process()

Оптимизация производительности ETL

• Parallel processing: используйте multiprocessing для Transform этапа
• Пакетные операции: загружайте данные блоками, а не по одной записи
• Индексы: убедитесь, что поля для поиска (SKU) проиндексированы
• Кэширование: кэшируйте справочники (категории, бренды)
• Monitoring: отслеживайте время выполнения каждого этапа

Для более сложных сценариев рассмотрите использование специализированных ETL-инструментов типа Apache Airflow, Luigi или готовых решений от Elbuz с поддержкой прямого подключения к базам данных поставщиков.

Обнаружение изменений и инкрементальные обновления

Эффективная стратегия обнаружения изменений (Change Detection) критически важна для минимизации нагрузки и ускорения синхронизации.

Стратегия 1: Timestamp-based Change Detection

Самый распространенный подход — использование полей с временем последнего изменения.

Requirements:

• Таблица должна иметь поле updated_at or modified_at
• Поле автоматически обновляется при изменении записи
• Индекс на поле timestamp для быстрого поиска

Реализация:

 SELECT * FROM products WHERE updated_at > '2025-10-21 10:00:00' ORDER BY updated_at ASC;

Advantages:

• Простота реализации
• Высокая производительность с индексом
• Надежность (если поле правильно обновляется)

Flaws:

• Не обнаруживает удаленные записи
• Зависимость от точности синхронизации часов
• Может пропустить изменения при откате времени

Стратегия 2: Checksum-based Change Detection

Вычисление хэша данных для определения изменений.

Реализация:

 import hashlib import json def calculate_checksum(product): """Вычисляет MD5 хэш критичных полей товара""" data ={ 'sku': product['sku'], 'name': product['name'], 'price': product['price'], 'stock': product['stock'] } # Создаем стабильную строку JSON json_str = json.dumps(data, sort_keys=True) # Вычисляем MD5 return hashlib.md5(json_str.encode()).hexdigest() def detect_changes(old_products, new_products): """Находит измененные товары""" changes ={ 'added': [], 'modified': [], 'deleted': [] } old_checksums ={p['sku']: calculate_checksum(p) for p in old_products} new_checksums ={p['sku']: calculate_checksum(p) for p in new_products} # Новые товары for sku in new_checksums: if sku not in old_checksums: changes['added'].append(sku) elif old_checksums[sku]!= new_checksums[sku]: changes['modified'].append(sku) # Удаленные товары for sku in old_checksums: if sku not in new_checksums: changes['deleted'].append(sku) return changes

Advantages:

• Обнаруживает любые изменения в данных
• Не зависит от timestamp полей
• Может обнаружить удаленные записи

Flaws:

• Требует загрузки всех данных для сравнения
• Дополнительные вычислительные затраты
• Сложность хранения checksums

Стратегия 3: Version-based Change Detection

Использование поля версии для отслеживания изменений.

Table structure:

 CREATE TABLE products ( id INT PRIMARY KEY, sku VARCHAR(50), name VARCHAR(255), price DECIMAL(10,2), version INT DEFAULT 1, updated_at TIMESTAMP ); -- Триггер автоматического увеличения версии CREATE TRIGGER increment_version BEFORE UPDATE ON products FOR EACH ROW SET NEW.version = OLD.version + 1;

Запрос измененных записей:

 SELECT * FROM products WHERE version > @last_known_version;

Стратегия 4: Change Data Capture (CDC)

Специализированные механизмы БД для отслеживания изменений.

MySQL Binary Log:

Использование binlog для отслеживания всех изменений.

• Требует прав на чтение binlog
• Позволяет получить все операции (INSERT, UPDATE, DELETE)
• Инструменты: Debezium, Maxwell

PostgreSQL Logical Replication:

Встроенный механизм репликации изменений.

• Настройка публикации на стороне поставщика
• Подписка на изменения на вашей стороне
• Реальное время без polling

Обработка удаленных записей

Удаленные товары не возвращаются в обычных запросах. Есть несколько подходов:

1. Soft Delete (рекомендуется)

Поставщик не удаляет записи физически, а помечает их.

 SELECT * FROM products WHERE (updated_at > @last_sync OR deleted_at > @last_sync) AND deleted_at IS NULL; -- Активные товары SELECT * FROM products WHERE deleted_at > @last_sync; -- Удаленные товары

2. Полное сравнение (для небольших каталогов)

 # Загрузить все текущие SKU current_skus = set(get_all_skus_from_supplier()) # Сравнить с локальными SKU local_skus = set(get_all_local_skus()) # Удаленные товары deleted_skus = local_skus - current_skus

3. Snapshot таблица

Поставщик создает snapshot-таблицу с актуальными товарами.

Оптимизация инкрементальных обновлений

1. Использование индексов

 -- Индекс на timestamp поле CREATE INDEX idx_products_updated ON products(updated_at); -- Композитный индекс для комплексных запросов CREATE INDEX idx_products_updated_active ON products(updated_at, active);

2. Партицирование таблиц

Для очень больших таблиц используйте партицирование по дате.

3. Материализованные представления

Предварительно подготовленные выборки для быстрого доступа.

Лучшая практика: Комбинируйте timestamp-based подход для обновлений и периодическое полное сравнение (раз в сутки/неделю) для выявления несоответствий и удаленных записей.

Резервное копирование и обработка сбоев

Прямое подключение к базе данных создает зависимость от доступности инфраструктуры поставщика. Необходим план на случай сбоев.

Стратегия резервного копирования

1. Кэширование данных поставщика

Храните локальную копию последних данных.

 import json from datetime import datetime def cache_supplier_data(products, cache_file='supplier_cache.json'): """Сохранение данных в кэш""" cache_data ={ 'timestamp': datetime.now().isoformat(), 'products': products } with open(cache_file, 'w', encoding='utf-8') as f: json.dump(cache_data, f, ensure_ascii=False, indent=2) def load_from_cache(cache_file='supplier_cache.json'): """Загрузка из кэша при недоступности БД""" try: with open(cache_file, 'r', encoding='utf-8') as f: cache_data = json.load(f) cache_age = datetime.now() - datetime.fromisoformat(cache_data['timestamp']) if cache_age.days > 1: logger.warning(f"Cache is {cache_age.days} days old") return cache_data['products'] except: return None

2. Fallback на файловый обмен

Резервный канал получения данных.

 def get_products(): """Получение данных с автоматическим fallback""" try: # Попытка 1: Прямое подключение к БД products = extract_from_database() cache_supplier_data(products) return products except DatabaseConnectionError as e: logger.error(f"Database connection failed: {e}") try: # Попытка 2: Загрузка файла-прайса logger.info("Trying to load from FTP...") products = load_from_ftp() return products except FTPError as e: logger.error(f"FTP fallback failed: {e}") # Попытка 3: Использование кэша logger.info("Loading from cache...") products = load_from_cache() if products: return products else: raise Exception("All data sources failed")

Обработка типовых ошибок

1. Ошибка подключения

 import time from mysql.connector import Error as MySQLError def connect_with_retry(max_retries=3, retry_delay=5): """Подключение с повторными попытками""" for attempt in range(max_retries): try: connection = mysql.connector.connect(**db_config) logger.info("Successfully connected to database") return connection except MySQLError as e: logger.warning(f"Connection attempt {attempt+1} failed: {e}") if attempt< max_retries - 1: logger.info(f"Retrying in {retry_delay} seconds...") time.sleep(retry_delay) else: logger.error("Max retries reached, connection failed") raise

2. Timeout запросов

 def execute_with_timeout(cursor, query, timeout=30): """Выполнение запроса с таймаутом""" cursor.execute("SET SESSION MAX_EXECUTION_TIME=%s", (timeout * 1000,)) try: cursor.execute(query) return cursor.fetchall() except Exception as e: logger.error(f"Query timeout or error: {e}") raise

3. Обработка разрыва соединения

 def robust_query(query, params=None): """Запрос с автоматическим переподключением""" connection = None max_attempts = 2 for attempt in range(max_attempts): try: if not connection or not connection.is_connected(): connection = mysql.connector.connect(**db_config) cursor = connection.cursor(dictionary=True) cursor.execute(query, params or ()) results = cursor.fetchall() cursor.close() return results except MySQLError as e: if attempt< max_attempts - 1 and is_connection_error(e): logger.warning("Connection lost, reconnecting...") connection = None continue else: raise finally: if connection and connection.is_connected(): connection.close() def is_connection_error(error): """Определение ошибок, требующих переподключения""" connection_errors = [2006, 2013, 2055] # MySQL error codes return error.errno in connection_errors

Мониторинг и алертинг

1. Health Check

 def check_database_health(): """Проверка доступности БД""" try: connection = mysql.connector.connect(**db_config) cursor = connection.cursor() cursor.execute("SELECT 1") cursor.fetchone() cursor.close() connection.close() return True except: return False def monitor_integration(): """Периодический мониторинг""" if not check_database_health(): send_alert("Database connection is down!") # Проверка давности последнего обновления last_update = get_last_update_time() if (datetime.now() - last_update).hours > 2: send_alert("No updates for 2 hours!")

2. Логирование всех операций

 import logging import logging.handlers def setup_logging(): logger = logging.getLogger('supplier_import') logger.setLevel(logging.INFO) # Файловый хендлер с ротацией file_handler = logging.handlers.RotatingFileHandler( 'supplier_import.log', maxBytes=10*1024*1024, # 10 MB backupCount=5 ) formatter = logging.Formatter( '%(asctime)s - %(name)s - %(levelname)s - %(message)s' ) file_handler.setFormatter(formatter) logger.addHandler(file_handler) return logger

3. Метрики производительности

 import time class ImportMetrics: def __init__(self): self.start_time = None self.metrics = {} def start(self): self.start_time = time.time() def record(self, name, value): self.metrics[name] = value def finish(self): duration = time.time() - self.start_time self.metrics['duration_seconds'] = duration return self.metrics # Использование metrics = ImportMetrics() metrics.start() products = extract_products() metrics.record('products_extracted', len(products)) transformed = transform_products(products) metrics.record('products_transformed', len(transformed)) loaded = load_products(transformed) metrics.record('products_loaded', loaded) result = metrics.finish() logger.info(f"Import metrics: {result}")

План действий при сбое

1. Detection: автоматический мониторинг или алерт от поставщика
2. Notification: отправка alert команде (email, Slack, SMS)
3. Fallback: переключение на резервный источник данных
4. Diagnostics: проверка логов, связь с поставщиком
5. Recovery: возврат к прямому подключению после устранения проблемы
6. Analysis: изучение причин сбоя для предотвращения в будущем

Important: Регулярно тестируйте процедуры восстановления. Не ждите реального сбоя — симулируйте отказ БД и убедитесь, что fallback механизмы работают корректно.

Frequently Asked Questions (FAQ)

Заключение и следующие шаги

Прямое подключение к базе данных поставщика — мощный инструмент для компаний, которым критически важна актуальность данных в реальном времени. Этот метод требует больше технических навыков и тщательной настройки безопасности по сравнению с файловым обменом, но преимущества значительны: нулевая задержка, контроль над выборкой данных, отсутствие проблем с форматами файлов.

Key findings:

• Безопасность прежде всего: используйте read-only доступ, VPN/SSH, IP whitelist и SSL
• Инкрементальные обновления: загружайте только изменения для минимизации нагрузки
• ETL процесс: структурируйте импорт на этапы Extract, Transform, Load
• Reservation: имейте план Б на случай недоступности БД поставщика
• Monitoring: отслеживайте все операции и настройте алерты
• Optimization: используйте индексы, пагинацию, пакетную обработку

Рекомендуемый порядок внедрения:

1. Переговоры с поставщиком и согласование доступа
2. Настройка безопасного подключения (VPN или SSH)
3. Тестовое подключение и изучение структуры данных
4. Разработка SQL-запросов для выборки
5. Реализация ETL-процесса с логированием
6. Тестирование на небольшом объеме данных
7. Настройка автоматизации (cron, планировщик)
8. Мониторинг и постепенное увеличение частоты обновлений

Альтернативные подходы:

Если прямое подключение к БД кажется слишком сложным, рассмотрите альтернативы:

• REST API интеграция — более безопасная и стандартизированная альтернатива
• Автоматическая загрузка файлов — проще в настройке, меньше зависимостей
• Ready-made solutions — платформы с встроенной поддержкой различных типов интеграций

Готовое решение от Elbuz

Если вам нужна система, которая поддерживает прямое подключение к базам данных поставщиков "из коробки", обратите внимание на Elbuz platform. Она включает:

• Готовые коннекторы для MySQL, PostgreSQL, MSSQL
• Визуальный конфигуратор подключений без программирования
• Встроенные механизмы безопасности и шифрования
• Автоматическое обнаружение изменений и инкрементальные обновления
• ETL-процессы с трансформацией и валидацией данных
• Мониторинг, алерты и детальное логирование
• Резервирование и fallback механизмы

Платформа позволяет настроить интеграцию за несколько часов вместо недель разработки. Узнайте больше о возможностях интеграции с базами данных или свяжитесь с нашей командой для консультации по вашему проекту.

Related articles

Для более глубокого понимания управления данными поставщиков рекомендуем изучить:

• Полное руководство по импорту и синхронизации прайс-листов — родительская pillar-статья
• Как настроить автоматическую синхронизацию цен и остатков — методы real-time синхронизации
• Автоматическая загрузка прайсов с почты, FTP или по ссылке — альтернативный подход
• XLS, CSV, XML, YML: руководство по форматам прайс-листов — понимание структур данных
• Как конвертировать прайс-листы между форматами — трансформация данных