update mongo for multi db support
This commit is contained in:
parent
4942574b94
commit
cbef304045
7 changed files with 927 additions and 13 deletions
212
docs/mongodb-multi-database-migration.md
Normal file
212
docs/mongodb-multi-database-migration.md
Normal file
|
|
@ -0,0 +1,212 @@
|
|||
# MongoDB Client Multi-Database Migration Guide
|
||||
|
||||
## Overview
|
||||
Your MongoDB client has been enhanced to support multiple databases dynamically while maintaining full backward compatibility.
|
||||
|
||||
## Key Features Added
|
||||
|
||||
### 1. **Dynamic Database Switching**
|
||||
```typescript
|
||||
// Set default database (all operations will use this unless overridden)
|
||||
client.setDefaultDatabase('analytics');
|
||||
|
||||
// Get current default database
|
||||
const currentDb = client.getDefaultDatabase(); // Returns: 'analytics'
|
||||
```
|
||||
|
||||
### 2. **Database Parameter in Methods**
|
||||
All methods now accept an optional `database` parameter:
|
||||
|
||||
```typescript
|
||||
// Old way (still works - uses default database)
|
||||
await client.batchUpsert('symbols', data, 'symbol');
|
||||
|
||||
// New way (specify database explicitly)
|
||||
await client.batchUpsert('symbols', data, 'symbol', { database: 'stock' });
|
||||
```
|
||||
|
||||
### 3. **Convenience Methods**
|
||||
Pre-configured methods for common databases:
|
||||
|
||||
```typescript
|
||||
// Stock database operations
|
||||
await client.batchUpsertStock('symbols', data, 'symbol');
|
||||
|
||||
// Analytics database operations
|
||||
await client.batchUpsertAnalytics('metrics', data, 'metric_name');
|
||||
|
||||
// Trading documents database operations
|
||||
await client.batchUpsertTrading('orders', data, 'order_id');
|
||||
```
|
||||
|
||||
### 4. **Direct Database Access**
|
||||
```typescript
|
||||
// Get specific database instances
|
||||
const stockDb = client.getDatabase('stock');
|
||||
const analyticsDb = client.getDatabase('analytics');
|
||||
|
||||
// Get collections with database override
|
||||
const collection = client.getCollection('symbols', 'stock');
|
||||
```
|
||||
|
||||
## Migration Steps
|
||||
|
||||
### Step 1: No Changes Required (Backward Compatible)
|
||||
Your existing code continues to work unchanged:
|
||||
|
||||
```typescript
|
||||
// This still works exactly as before
|
||||
const client = MongoDBClient.getInstance();
|
||||
await client.connect();
|
||||
await client.batchUpsert('exchanges', exchangeData, 'exchange_id');
|
||||
```
|
||||
|
||||
### Step 2: Organize Data by Database (Recommended)
|
||||
Update your data service to use appropriate databases:
|
||||
|
||||
```typescript
|
||||
// In your data service initialization
|
||||
export class DataService {
|
||||
private mongoClient = MongoDBClient.getInstance();
|
||||
|
||||
async initialize() {
|
||||
await this.mongoClient.connect();
|
||||
|
||||
// Set stock as default for most operations
|
||||
this.mongoClient.setDefaultDatabase('stock');
|
||||
}
|
||||
|
||||
async saveInteractiveBrokersData(exchanges: any[], symbols: any[]) {
|
||||
// Stock market data goes to 'stock' database (default)
|
||||
await this.mongoClient.batchUpsert('exchanges', exchanges, 'exchange_id');
|
||||
await this.mongoClient.batchUpsert('symbols', symbols, 'symbol');
|
||||
}
|
||||
|
||||
async saveAnalyticsData(performance: any[]) {
|
||||
// Analytics data goes to 'analytics' database
|
||||
await this.mongoClient.batchUpsert(
|
||||
'performance',
|
||||
performance,
|
||||
'date',
|
||||
{ database: 'analytics' }
|
||||
);
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
### Step 3: Use Convenience Methods (Optional)
|
||||
Replace explicit database parameters with convenience methods:
|
||||
|
||||
```typescript
|
||||
// Instead of:
|
||||
await client.batchUpsert('symbols', data, 'symbol', { database: 'stock' });
|
||||
|
||||
// Use:
|
||||
await client.batchUpsertStock('symbols', data, 'symbol');
|
||||
```
|
||||
|
||||
## Factory Functions
|
||||
New factory functions are available for easier database management:
|
||||
|
||||
```typescript
|
||||
import {
|
||||
connectMongoDB,
|
||||
setDefaultDatabase,
|
||||
getCurrentDatabase,
|
||||
getDatabase
|
||||
} from '@stock-bot/mongodb-client';
|
||||
|
||||
// Set default database globally
|
||||
setDefaultDatabase('analytics');
|
||||
|
||||
// Get current default
|
||||
const current = getCurrentDatabase();
|
||||
|
||||
// Get specific database
|
||||
const stockDb = getDatabase('stock');
|
||||
```
|
||||
|
||||
## Database Recommendations
|
||||
|
||||
### Stock Database (`stock`)
|
||||
- Market data (symbols, exchanges, prices)
|
||||
- Financial instruments
|
||||
- Market events
|
||||
- Real-time data
|
||||
|
||||
### Analytics Database (`analytics`)
|
||||
- Performance metrics
|
||||
- Calculated indicators
|
||||
- Reports and dashboards
|
||||
- Aggregated data
|
||||
|
||||
### Trading Documents Database (`trading_documents`)
|
||||
- Trade orders and executions
|
||||
- User portfolios
|
||||
- Transaction logs
|
||||
- Audit trails
|
||||
|
||||
## Example: Updating Your Data Service
|
||||
|
||||
```typescript
|
||||
// Before (still works)
|
||||
export class DataService {
|
||||
async saveExchanges(exchanges: any[]) {
|
||||
const client = MongoDBClient.getInstance();
|
||||
await client.batchUpsert('exchanges', exchanges, 'exchange_id');
|
||||
}
|
||||
}
|
||||
|
||||
// After (recommended)
|
||||
export class DataService {
|
||||
private mongoClient = MongoDBClient.getInstance();
|
||||
|
||||
async initialize() {
|
||||
await this.mongoClient.connect();
|
||||
this.mongoClient.setDefaultDatabase('stock'); // Set appropriate default
|
||||
}
|
||||
|
||||
async saveExchanges(exchanges: any[]) {
|
||||
// Uses default 'stock' database
|
||||
await this.mongoClient.batchUpsert('exchanges', exchanges, 'exchange_id');
|
||||
|
||||
// Or use convenience method
|
||||
await this.mongoClient.batchUpsertStock('exchanges', exchanges, 'exchange_id');
|
||||
}
|
||||
|
||||
async savePerformanceMetrics(metrics: any[]) {
|
||||
// Save to analytics database
|
||||
await this.mongoClient.batchUpsertAnalytics('metrics', metrics, 'metric_name');
|
||||
}
|
||||
}
|
||||
```
|
||||
|
||||
## Testing
|
||||
Your existing tests continue to work. For new multi-database features:
|
||||
|
||||
```typescript
|
||||
import { MongoDBClient } from '@stock-bot/mongodb-client';
|
||||
|
||||
const client = MongoDBClient.getInstance();
|
||||
await client.connect();
|
||||
|
||||
// Test database switching
|
||||
client.setDefaultDatabase('test_db');
|
||||
expect(client.getDefaultDatabase()).toBe('test_db');
|
||||
|
||||
// Test explicit database parameter
|
||||
await client.batchUpsert('test_collection', data, 'id', { database: 'other_db' });
|
||||
```
|
||||
|
||||
## Benefits
|
||||
1. **Organized Data**: Separate databases for different data types
|
||||
2. **Better Performance**: Smaller, focused databases
|
||||
3. **Easier Maintenance**: Clear data boundaries
|
||||
4. **Scalability**: Can scale databases independently
|
||||
5. **Backward Compatibility**: No breaking changes
|
||||
|
||||
## Next Steps
|
||||
1. Update your data service to use appropriate default database
|
||||
2. Gradually migrate to using specific databases for different data types
|
||||
3. Consider using convenience methods for cleaner code
|
||||
4. Update tests to cover multi-database scenarios
|
||||
216
examples/enhanced-data-service.ts
Normal file
216
examples/enhanced-data-service.ts
Normal file
|
|
@ -0,0 +1,216 @@
|
|||
/**
|
||||
* Example: Enhanced Data Service with Multi-Database Support
|
||||
*
|
||||
* This shows how to update your existing data service to leverage
|
||||
* the new multi-database MongoDB client functionality.
|
||||
*/
|
||||
|
||||
import { getLogger } from '@stock-bot/logger';
|
||||
import { MongoDBClient, setDefaultDatabase } from '@stock-bot/mongodb-client';
|
||||
|
||||
const logger = getLogger('enhanced-data-service');
|
||||
|
||||
export class EnhancedDataService {
|
||||
private mongoClient = MongoDBClient.getInstance();
|
||||
|
||||
async initialize() {
|
||||
// Connect to MongoDB
|
||||
await this.mongoClient.connect();
|
||||
|
||||
// Set stock as default database for market data operations
|
||||
setDefaultDatabase('stock');
|
||||
logger.info('Enhanced data service initialized with multi-database support');
|
||||
}
|
||||
|
||||
/**
|
||||
* Save Interactive Brokers data to stock database
|
||||
*/
|
||||
async saveIBMarketData(exchanges: any[], symbols: any[]) {
|
||||
logger.info('Saving IB market data to stock database');
|
||||
|
||||
// These use the default 'stock' database
|
||||
const exchangeResult = await this.mongoClient.batchUpsert(
|
||||
'exchanges',
|
||||
exchanges,
|
||||
'exchange_id'
|
||||
);
|
||||
|
||||
const symbolResult = await this.mongoClient.batchUpsert('symbols', symbols, 'symbol');
|
||||
|
||||
// Or use convenience method for cleaner code
|
||||
// const exchangeResult = await this.mongoClient.batchUpsertStock('exchanges', exchanges, 'exchange_id');
|
||||
// const symbolResult = await this.mongoClient.batchUpsertStock('symbols', symbols, 'symbol');
|
||||
|
||||
logger.info('IB market data saved', {
|
||||
exchanges: exchangeResult,
|
||||
symbols: symbolResult,
|
||||
});
|
||||
|
||||
return { exchanges: exchangeResult, symbols: symbolResult };
|
||||
}
|
||||
|
||||
/**
|
||||
* Save real-time prices to stock database
|
||||
*/
|
||||
async saveRealTimePrices(priceData: any[]) {
|
||||
logger.info(`Saving ${priceData.length} real-time prices to stock database`);
|
||||
|
||||
return await this.mongoClient.batchUpsertStock('real_time_prices', priceData, [
|
||||
'symbol',
|
||||
'timestamp',
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Save performance analytics to analytics database
|
||||
*/
|
||||
async savePerformanceAnalytics(performanceData: any[]) {
|
||||
logger.info(`Saving ${performanceData.length} performance records to analytics database`);
|
||||
|
||||
return await this.mongoClient.batchUpsertAnalytics('performance_metrics', performanceData, [
|
||||
'portfolio_id',
|
||||
'date',
|
||||
]);
|
||||
}
|
||||
|
||||
/**
|
||||
* Save trading logs to trading_documents database
|
||||
*/
|
||||
async saveTradingLogs(logs: any[]) {
|
||||
logger.info(`Saving ${logs.length} trading logs to trading_documents database`);
|
||||
|
||||
return await this.mongoClient.batchUpsertTrading('trading_logs', logs, 'log_id');
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Cross-database analytics
|
||||
*/
|
||||
async generateCrossDatabaseReport() {
|
||||
logger.info('Generating cross-database analytics report');
|
||||
|
||||
// Get data from multiple databases
|
||||
const stockDb = this.mongoClient.getDatabase('stock');
|
||||
const analyticsDb = this.mongoClient.getDatabase('analytics');
|
||||
const tradingDb = this.mongoClient.getDatabase('trading_documents');
|
||||
|
||||
// Get counts from different databases
|
||||
const symbolCount = await stockDb.collection('symbols').countDocuments();
|
||||
const exchangeCount = await stockDb.collection('exchanges').countDocuments();
|
||||
const performanceCount = await analyticsDb.collection('performance_metrics').countDocuments();
|
||||
const logCount = await tradingDb.collection('trading_logs').countDocuments();
|
||||
|
||||
const report = {
|
||||
stock_database: {
|
||||
symbols: symbolCount,
|
||||
exchanges: exchangeCount,
|
||||
},
|
||||
analytics_database: {
|
||||
performance_metrics: performanceCount,
|
||||
},
|
||||
trading_database: {
|
||||
logs: logCount,
|
||||
},
|
||||
generated_at: new Date(),
|
||||
};
|
||||
|
||||
logger.info('Cross-database report generated', report);
|
||||
return report;
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Database-specific operations
|
||||
*/
|
||||
async performDatabaseSpecificOperations() {
|
||||
// Switch default database for a series of analytics operations
|
||||
const originalDefault = this.mongoClient.getDefaultDatabase();
|
||||
this.mongoClient.setDefaultDatabase('analytics');
|
||||
|
||||
try {
|
||||
// Now all operations without explicit database go to 'analytics'
|
||||
await this.mongoClient.batchUpsert('daily_reports', [], 'date');
|
||||
await this.mongoClient.batchUpsert('portfolio_summaries', [], 'portfolio_id');
|
||||
|
||||
logger.info('Analytics operations completed on analytics database');
|
||||
} finally {
|
||||
// Restore original default database
|
||||
this.mongoClient.setDefaultDatabase(originalDefault);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Example: Configuration-driven database routing
|
||||
*/
|
||||
async saveDataByType(dataType: string, collection: string, data: any[], uniqueKeys: string[]) {
|
||||
const databaseMapping: Record<string, string> = {
|
||||
market_data: 'stock',
|
||||
analytics: 'analytics',
|
||||
trading: 'trading_documents',
|
||||
logs: 'trading_documents',
|
||||
};
|
||||
|
||||
const targetDatabase = databaseMapping[dataType] || 'stock';
|
||||
|
||||
logger.info(`Saving ${data.length} records`, {
|
||||
dataType,
|
||||
targetDatabase,
|
||||
collection,
|
||||
});
|
||||
|
||||
return await this.mongoClient.batchUpsert(collection, data, uniqueKeys, {
|
||||
database: targetDatabase,
|
||||
});
|
||||
}
|
||||
|
||||
async shutdown() {
|
||||
logger.info('Shutting down enhanced data service');
|
||||
await this.mongoClient.disconnect();
|
||||
}
|
||||
}
|
||||
|
||||
// Usage example for your existing data service
|
||||
export async function integrateWithExistingDataService() {
|
||||
const enhancedService = new EnhancedDataService();
|
||||
|
||||
try {
|
||||
await enhancedService.initialize();
|
||||
|
||||
// Example market data
|
||||
const exchanges = [
|
||||
{ exchange_id: 'NYSE', name: 'New York Stock Exchange', mic: 'XNYS' },
|
||||
{ exchange_id: 'NASDAQ', name: 'NASDAQ', mic: 'XNAS' },
|
||||
];
|
||||
|
||||
const symbols = [
|
||||
{ symbol: 'AAPL', exchange: 'NASDAQ', company_name: 'Apple Inc.' },
|
||||
{ symbol: 'MSFT', exchange: 'NASDAQ', company_name: 'Microsoft Corporation' },
|
||||
];
|
||||
|
||||
// Save to appropriate databases
|
||||
await enhancedService.saveIBMarketData(exchanges, symbols);
|
||||
|
||||
// Save real-time prices
|
||||
await enhancedService.saveRealTimePrices([
|
||||
{ symbol: 'AAPL', price: 150.25, volume: 1000000, timestamp: new Date() },
|
||||
]);
|
||||
|
||||
// Save analytics
|
||||
await enhancedService.savePerformanceAnalytics([
|
||||
{ portfolio_id: 'portfolio_1', date: new Date(), return: 0.15 },
|
||||
]);
|
||||
|
||||
// Generate cross-database report
|
||||
const report = await enhancedService.generateCrossDatabaseReport();
|
||||
console.log('Cross-database report:', report);
|
||||
|
||||
// Example of configuration-driven routing
|
||||
await enhancedService.saveDataByType('market_data', 'prices', [], 'symbol');
|
||||
await enhancedService.saveDataByType('analytics', 'metrics', [], 'metric_id');
|
||||
} catch (error) {
|
||||
logger.error('Enhanced data service error:', error);
|
||||
} finally {
|
||||
await enhancedService.shutdown();
|
||||
}
|
||||
}
|
||||
|
||||
// Export for integration
|
||||
export default EnhancedDataService;
|
||||
230
examples/mongodb-multi-database-usage.ts
Normal file
230
examples/mongodb-multi-database-usage.ts
Normal file
|
|
@ -0,0 +1,230 @@
|
|||
/**
|
||||
* Practical Usage Examples for Multi-Database MongoDB Client
|
||||
*
|
||||
* This file demonstrates real-world usage patterns for the enhanced MongoDB client
|
||||
* with multiple database support.
|
||||
*/
|
||||
|
||||
import { getCurrentDatabase, MongoDBClient, setDefaultDatabase } from '@stock-bot/mongodb-client';
|
||||
|
||||
// Example 1: Using different databases for different data types
|
||||
export class DataServiceExample {
|
||||
private mongoClient = MongoDBClient.getInstance();
|
||||
|
||||
async initialize() {
|
||||
await this.mongoClient.connect();
|
||||
|
||||
// Set stock as default database for most operations
|
||||
setDefaultDatabase('stock');
|
||||
console.log(`Default database: ${getCurrentDatabase()}`);
|
||||
}
|
||||
|
||||
// Stock market data goes to 'stock' database (default)
|
||||
async saveStockData(symbols: any[], exchanges: any[]) {
|
||||
// These use the default 'stock' database
|
||||
await this.mongoClient.batchUpsert('symbols', symbols, 'symbol');
|
||||
await this.mongoClient.batchUpsert('exchanges', exchanges, 'exchange_id');
|
||||
|
||||
// Or use convenience method (explicitly targets 'stock' database)
|
||||
await this.mongoClient.batchUpsertStock('prices', symbols, 'symbol');
|
||||
}
|
||||
|
||||
// Analytics and metrics go to 'analytics' database
|
||||
async saveAnalyticsData(performanceData: any[], metrics: any[]) {
|
||||
// Override database for specific operations
|
||||
await this.mongoClient.batchUpsert('performance', performanceData, 'date', {
|
||||
database: 'analytics',
|
||||
});
|
||||
|
||||
// Or use convenience method
|
||||
await this.mongoClient.batchUpsertAnalytics('metrics', metrics, 'metric_name');
|
||||
}
|
||||
|
||||
// Trading documents and logs go to 'trading_documents' database
|
||||
async saveTradingData(orders: any[], transactions: any[]) {
|
||||
// Use convenience method for trading data
|
||||
await this.mongoClient.batchUpsertTrading('orders', orders, 'order_id');
|
||||
await this.mongoClient.batchUpsertTrading('transactions', transactions, 'transaction_id');
|
||||
}
|
||||
|
||||
// Example of switching default database dynamically
|
||||
async switchToAnalyticsMode() {
|
||||
console.log(`Current default: ${getCurrentDatabase()}`);
|
||||
|
||||
// Switch to analytics database for a series of operations
|
||||
setDefaultDatabase('analytics');
|
||||
console.log(`New default: ${getCurrentDatabase()}`);
|
||||
|
||||
// Now all operations without explicit database parameter go to 'analytics'
|
||||
await this.mongoClient.batchUpsert('daily_reports', [], 'date');
|
||||
await this.mongoClient.batchUpsert('portfolio_performance', [], 'portfolio_id');
|
||||
|
||||
// Switch back to stock database
|
||||
setDefaultDatabase('stock');
|
||||
}
|
||||
|
||||
// Example of working with multiple databases simultaneously
|
||||
async crossDatabaseAnalysis() {
|
||||
// Get direct access to different databases
|
||||
const stockDb = this.mongoClient.getDatabase('stock');
|
||||
const analyticsDb = this.mongoClient.getDatabase('analytics');
|
||||
const tradingDb = this.mongoClient.getDatabase('trading_documents');
|
||||
|
||||
// Perform operations on multiple databases
|
||||
const stockSymbols = await stockDb.collection('symbols').find({}).toArray();
|
||||
const performance = await analyticsDb.collection('performance').find({}).toArray();
|
||||
const orders = await tradingDb.collection('orders').find({}).toArray();
|
||||
|
||||
console.log('Cross-database analysis:', {
|
||||
symbolsCount: stockSymbols.length,
|
||||
performanceRecords: performance.length,
|
||||
ordersCount: orders.length,
|
||||
});
|
||||
}
|
||||
}
|
||||
|
||||
// Example 2: Data Migration Between Databases
|
||||
export class DataMigrationExample {
|
||||
private mongoClient = MongoDBClient.getInstance();
|
||||
|
||||
async migrateHistoricalData() {
|
||||
await this.mongoClient.connect();
|
||||
|
||||
// Get collections from different databases
|
||||
const stockCollection = this.mongoClient.getCollection('historical_prices', 'stock');
|
||||
const analyticsCollection = this.mongoClient.getCollection('price_analysis', 'analytics');
|
||||
|
||||
// Read from stock database
|
||||
const historicalPrices = await stockCollection
|
||||
.find({
|
||||
date: { $gte: new Date('2024-01-01') },
|
||||
})
|
||||
.toArray();
|
||||
|
||||
console.log(`Found ${historicalPrices.length} historical price records`);
|
||||
|
||||
// Transform and save to analytics database
|
||||
const analysisData = historicalPrices.map(price => ({
|
||||
symbol: price.symbol,
|
||||
date: price.date,
|
||||
price_change: price.close - price.open,
|
||||
volume_normalized: price.volume / 1000000,
|
||||
created_at: new Date(),
|
||||
updated_at: new Date(),
|
||||
}));
|
||||
|
||||
// Save to analytics database
|
||||
await this.mongoClient.batchUpsert('price_analysis', analysisData, ['symbol', 'date'], {
|
||||
database: 'analytics',
|
||||
});
|
||||
|
||||
console.log(`Migrated ${analysisData.length} records to analytics database`);
|
||||
}
|
||||
}
|
||||
|
||||
// Example 3: Service-Specific Database Usage
|
||||
export class TradingServiceExample {
|
||||
private mongoClient = MongoDBClient.getInstance();
|
||||
|
||||
async initialize() {
|
||||
await this.mongoClient.connect();
|
||||
// Trading service primarily works with trading_documents database
|
||||
setDefaultDatabase('trading_documents');
|
||||
}
|
||||
|
||||
async processTradeOrders(orders: any[]) {
|
||||
// Default database is 'trading_documents', so no need to specify
|
||||
await this.mongoClient.batchUpsert('orders', orders, 'order_id');
|
||||
|
||||
// Log to analytics database for monitoring
|
||||
const orderMetrics = orders.map(order => ({
|
||||
metric_name: `order_${order.type}`,
|
||||
value: order.quantity,
|
||||
timestamp: new Date(),
|
||||
}));
|
||||
|
||||
await this.mongoClient.batchUpsert(
|
||||
'trading_metrics',
|
||||
orderMetrics,
|
||||
['metric_name', 'timestamp'],
|
||||
{ database: 'analytics' }
|
||||
);
|
||||
}
|
||||
|
||||
async getOrderHistory(symbolFilter?: string) {
|
||||
// Get collection from default database (trading_documents)
|
||||
const ordersCollection = this.mongoClient.getCollection('orders');
|
||||
|
||||
const filter = symbolFilter ? { symbol: symbolFilter } : {};
|
||||
return await ordersCollection.find(filter).sort({ created_at: -1 }).limit(100).toArray();
|
||||
}
|
||||
}
|
||||
|
||||
// Example 4: Configuration-Based Database Routing
|
||||
export class ConfigurableDatabaseRouter {
|
||||
private mongoClient = MongoDBClient.getInstance();
|
||||
|
||||
// Configuration mapping data types to databases
|
||||
private databaseConfig = {
|
||||
market_data: 'stock',
|
||||
user_data: 'trading_documents',
|
||||
analytics: 'analytics',
|
||||
logs: 'trading_documents',
|
||||
cache: 'analytics',
|
||||
};
|
||||
|
||||
async saveData(
|
||||
dataType: keyof typeof this.databaseConfig,
|
||||
collection: string,
|
||||
data: any[],
|
||||
uniqueKeys: string[]
|
||||
) {
|
||||
const targetDatabase = this.databaseConfig[dataType];
|
||||
|
||||
console.log(`Saving ${data.length} records to ${targetDatabase}.${collection}`);
|
||||
|
||||
return await this.mongoClient.batchUpsert(collection, data, uniqueKeys, {
|
||||
database: targetDatabase,
|
||||
});
|
||||
}
|
||||
|
||||
async saveMarketData(data: any[]) {
|
||||
return this.saveData('market_data', 'realtime_prices', data, 'symbol');
|
||||
}
|
||||
|
||||
async saveUserActivity(data: any[]) {
|
||||
return this.saveData('user_data', 'user_actions', data, 'user_id');
|
||||
}
|
||||
|
||||
async saveAnalytics(data: any[]) {
|
||||
return this.saveData('analytics', 'performance_metrics', data, 'metric_id');
|
||||
}
|
||||
}
|
||||
|
||||
// Usage example for your data service
|
||||
export async function exampleUsage() {
|
||||
const dataService = new DataServiceExample();
|
||||
|
||||
await dataService.initialize();
|
||||
|
||||
// Save different types of data to appropriate databases
|
||||
await dataService.saveStockData(
|
||||
[{ symbol: 'AAPL', price: 150.25, volume: 1000000 }],
|
||||
[{ exchange_id: 'NYSE', name: 'New York Stock Exchange' }]
|
||||
);
|
||||
|
||||
await dataService.saveAnalyticsData(
|
||||
[{ date: new Date(), portfolio_return: 0.15 }],
|
||||
[{ metric_name: 'sharpe_ratio', value: 1.25 }]
|
||||
);
|
||||
|
||||
await dataService.saveTradingData(
|
||||
[{ order_id: 'ORD001', symbol: 'AAPL', quantity: 100 }],
|
||||
[{ transaction_id: 'TXN001', amount: 15025 }]
|
||||
);
|
||||
|
||||
// Perform cross-database analysis
|
||||
await dataService.crossDatabaseAnalysis();
|
||||
|
||||
console.log('Multi-database operations completed successfully!');
|
||||
}
|
||||
|
|
@ -13,6 +13,7 @@ export class MongoDBClient {
|
|||
private static instance: MongoDBClient | null = null;
|
||||
private client: MongoClient | null = null;
|
||||
private db: Db | null = null;
|
||||
private defaultDatabase: string = 'stock'; // Default database name
|
||||
private readonly logger = getLogger('mongodb-client-simple');
|
||||
private isConnected = false;
|
||||
|
||||
|
|
@ -51,7 +52,9 @@ export class MongoDBClient {
|
|||
await this.client.connect();
|
||||
await this.client.db(mongodbConfig.MONGODB_DATABASE).admin().ping();
|
||||
|
||||
this.db = this.client.db(mongodbConfig.MONGODB_DATABASE);
|
||||
// Set default database from config
|
||||
this.defaultDatabase = mongodbConfig.MONGODB_DATABASE;
|
||||
this.db = this.client.db(this.defaultDatabase);
|
||||
this.isConnected = true;
|
||||
|
||||
this.logger.info('Successfully connected to MongoDB');
|
||||
|
|
@ -85,6 +88,36 @@ export class MongoDBClient {
|
|||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the default database for operations
|
||||
*/
|
||||
setDefaultDatabase(databaseName: string): void {
|
||||
this.defaultDatabase = databaseName;
|
||||
if (this.client) {
|
||||
this.db = this.client.db(databaseName);
|
||||
this.logger.info(`Default database changed to: ${databaseName}`);
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current default database name
|
||||
*/
|
||||
getDefaultDatabase(): string {
|
||||
return this.defaultDatabase;
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a database instance by name
|
||||
*/
|
||||
getDatabase(databaseName?: string): Db {
|
||||
if (!this.client) {
|
||||
throw new Error('MongoDB client not connected');
|
||||
}
|
||||
|
||||
const dbName = databaseName || this.defaultDatabase;
|
||||
return this.client.db(dbName);
|
||||
}
|
||||
|
||||
/**
|
||||
* Batch upsert documents for high-performance operations
|
||||
* Supports single or multiple unique keys for matching
|
||||
|
|
@ -97,9 +130,10 @@ export class MongoDBClient {
|
|||
uniqueKeys: string | string[],
|
||||
options: {
|
||||
chunkSize?: number;
|
||||
database?: string; // Optional database override
|
||||
} = {}
|
||||
): Promise<{ insertedCount: number; updatedCount: number; errors: unknown[] }> {
|
||||
if (!this.db) {
|
||||
if (!this.client) {
|
||||
throw new Error('MongoDB client not connected');
|
||||
}
|
||||
|
||||
|
|
@ -114,15 +148,18 @@ export class MongoDBClient {
|
|||
throw new Error('At least one unique key must be provided');
|
||||
}
|
||||
|
||||
const { chunkSize = 10000 } = options;
|
||||
const collection = this.db.collection<T>(collectionName);
|
||||
const { chunkSize = 10000, database } = options;
|
||||
const db = this.getDatabase(database);
|
||||
const collection = db.collection<T>(collectionName);
|
||||
const operationId = Math.random().toString(36).substring(7);
|
||||
const dbName = database || this.defaultDatabase;
|
||||
|
||||
let totalInserted = 0;
|
||||
let totalUpdated = 0;
|
||||
const errors: unknown[] = [];
|
||||
|
||||
this.logger.info(`Starting batch upsert operation [${operationId}]`, {
|
||||
database: dbName,
|
||||
collection: collectionName,
|
||||
totalDocuments: documents.length,
|
||||
uniqueKeys: keyFields,
|
||||
|
|
@ -186,11 +223,13 @@ export class MongoDBClient {
|
|||
inserted,
|
||||
updated,
|
||||
executionTime,
|
||||
database: dbName,
|
||||
collection: collectionName,
|
||||
});
|
||||
} catch (error) {
|
||||
this.logger.error(`Batch upsert failed on chunk [${operationId}]`, {
|
||||
error,
|
||||
database: dbName,
|
||||
collection: collectionName,
|
||||
chunkNumber: Math.floor(i / chunkSize) + 1,
|
||||
chunkStart: i,
|
||||
|
|
@ -202,6 +241,7 @@ export class MongoDBClient {
|
|||
}
|
||||
|
||||
this.logger.info(`Batch upsert completed [${operationId}]`, {
|
||||
database: dbName,
|
||||
collection: collectionName,
|
||||
totalRecords: documents.length,
|
||||
inserted: totalInserted,
|
||||
|
|
@ -216,11 +256,9 @@ export class MongoDBClient {
|
|||
/**
|
||||
* Get a typed collection
|
||||
*/
|
||||
getCollection<T extends DocumentBase>(name: string): Collection<T> {
|
||||
if (!this.db) {
|
||||
throw new Error('MongoDB client not connected');
|
||||
}
|
||||
return this.db.collection<T>(name);
|
||||
getCollection<T extends DocumentBase>(name: string, database?: string): Collection<T> {
|
||||
const db = this.getDatabase(database);
|
||||
return db.collection<T>(name);
|
||||
}
|
||||
|
||||
/**
|
||||
|
|
@ -229,9 +267,10 @@ export class MongoDBClient {
|
|||
async insertOne<T extends DocumentBase>(
|
||||
collectionName: string,
|
||||
document: Omit<T, '_id' | 'created_at' | 'updated_at'> &
|
||||
Partial<Pick<T, 'created_at' | 'updated_at'>>
|
||||
Partial<Pick<T, 'created_at' | 'updated_at'>>,
|
||||
database?: string
|
||||
): Promise<T> {
|
||||
const collection = this.getCollection<T>(collectionName);
|
||||
const collection = this.getCollection<T>(collectionName, database);
|
||||
|
||||
const now = new Date();
|
||||
const docWithTimestamps = {
|
||||
|
|
@ -252,12 +291,61 @@ export class MongoDBClient {
|
|||
}
|
||||
|
||||
/**
|
||||
* Get the database instance
|
||||
* Get the default database instance
|
||||
*/
|
||||
get database(): Db | null {
|
||||
return this.db;
|
||||
}
|
||||
|
||||
/**
|
||||
* Convenience methods for common databases
|
||||
*/
|
||||
|
||||
// Stock database operations
|
||||
async batchUpsertStock<T extends DocumentBase>(
|
||||
collectionName: string,
|
||||
documents: Array<
|
||||
Omit<T, '_id' | 'created_at' | 'updated_at'> & Partial<Pick<T, 'created_at' | 'updated_at'>>
|
||||
>,
|
||||
uniqueKeys: string | string[],
|
||||
options: { chunkSize?: number } = {}
|
||||
) {
|
||||
return this.batchUpsert(collectionName, documents, uniqueKeys, {
|
||||
...options,
|
||||
database: 'stock',
|
||||
});
|
||||
}
|
||||
|
||||
// Trading documents database operations
|
||||
async batchUpsertTrading<T extends DocumentBase>(
|
||||
collectionName: string,
|
||||
documents: Array<
|
||||
Omit<T, '_id' | 'created_at' | 'updated_at'> & Partial<Pick<T, 'created_at' | 'updated_at'>>
|
||||
>,
|
||||
uniqueKeys: string | string[],
|
||||
options: { chunkSize?: number } = {}
|
||||
) {
|
||||
return this.batchUpsert(collectionName, documents, uniqueKeys, {
|
||||
...options,
|
||||
database: 'trading_documents',
|
||||
});
|
||||
}
|
||||
|
||||
// Analytics database operations
|
||||
async batchUpsertAnalytics<T extends DocumentBase>(
|
||||
collectionName: string,
|
||||
documents: Array<
|
||||
Omit<T, '_id' | 'created_at' | 'updated_at'> & Partial<Pick<T, 'created_at' | 'updated_at'>>
|
||||
>,
|
||||
uniqueKeys: string | string[],
|
||||
options: { chunkSize?: number } = {}
|
||||
) {
|
||||
return this.batchUpsert(collectionName, documents, uniqueKeys, {
|
||||
...options,
|
||||
database: 'analytics',
|
||||
});
|
||||
}
|
||||
|
||||
private buildConnectionUri(): string {
|
||||
if (mongodbConfig.MONGODB_URI) {
|
||||
return mongodbConfig.MONGODB_URI;
|
||||
|
|
|
|||
|
|
@ -27,3 +27,27 @@ export async function disconnectMongoDB(): Promise<void> {
|
|||
await client.disconnect();
|
||||
}
|
||||
}
|
||||
|
||||
/**
|
||||
* Set the default database for all operations
|
||||
*/
|
||||
export function setDefaultDatabase(databaseName: string): void {
|
||||
const client = getMongoDBClient();
|
||||
client.setDefaultDatabase(databaseName);
|
||||
}
|
||||
|
||||
/**
|
||||
* Get the current default database name
|
||||
*/
|
||||
export function getCurrentDatabase(): string {
|
||||
const client = getMongoDBClient();
|
||||
return client.getDefaultDatabase();
|
||||
}
|
||||
|
||||
/**
|
||||
* Get a database instance by name
|
||||
*/
|
||||
export function getDatabase(databaseName?: string) {
|
||||
const client = getMongoDBClient();
|
||||
return client.getDatabase(databaseName);
|
||||
}
|
||||
|
|
|
|||
|
|
@ -19,4 +19,11 @@ export type {
|
|||
} from './types';
|
||||
|
||||
// Factory functions
|
||||
export { connectMongoDB, disconnectMongoDB, getMongoDBClient } from './factory';
|
||||
export {
|
||||
connectMongoDB,
|
||||
disconnectMongoDB,
|
||||
getCurrentDatabase,
|
||||
getDatabase,
|
||||
getMongoDBClient,
|
||||
setDefaultDatabase,
|
||||
} from './factory';
|
||||
|
|
|
|||
137
test-mongodb-multi-db.ts
Normal file
137
test-mongodb-multi-db.ts
Normal file
|
|
@ -0,0 +1,137 @@
|
|||
#!/usr/bin/env bun
|
||||
/**
|
||||
* Test script for MongoDB Client with Multiple Database Support
|
||||
*/
|
||||
import { MongoDBClient } from './libs/mongodb-client/src/client';
|
||||
|
||||
interface TestDocument {
|
||||
_id?: string;
|
||||
name: string;
|
||||
value: number;
|
||||
created_at?: Date;
|
||||
updated_at?: Date;
|
||||
}
|
||||
|
||||
async function testMultiDatabaseSupport() {
|
||||
const client = MongoDBClient.getInstance();
|
||||
|
||||
try {
|
||||
console.log('🔌 Connecting to MongoDB...');
|
||||
await client.connect();
|
||||
console.log('✅ Connected successfully!');
|
||||
|
||||
// Test 1: Check default database
|
||||
console.log('\n📊 Testing Default Database Operations');
|
||||
console.log(`Default database: ${client.getDefaultDatabase()}`);
|
||||
|
||||
// Test 2: Insert into default database (stock)
|
||||
const stockData: TestDocument[] = [
|
||||
{ name: 'AAPL', value: 150.25 },
|
||||
{ name: 'GOOGL', value: 2750.8 },
|
||||
{ name: 'MSFT', value: 305.15 },
|
||||
];
|
||||
|
||||
console.log('\n💾 Inserting into default database (stock)...');
|
||||
const stockResult = await client.batchUpsert('test_symbols', stockData, 'name');
|
||||
console.log('Stock database result:', stockResult);
|
||||
|
||||
// Test 3: Change default database
|
||||
console.log('\n🔄 Changing default database to analytics...');
|
||||
client.setDefaultDatabase('analytics');
|
||||
console.log(`New default database: ${client.getDefaultDatabase()}`);
|
||||
|
||||
// Test 4: Insert into new default database
|
||||
const analyticsData: TestDocument[] = [
|
||||
{ name: 'daily_volume', value: 1000000 },
|
||||
{ name: 'avg_price', value: 125.5 },
|
||||
{ name: 'volatility', value: 0.25 },
|
||||
];
|
||||
|
||||
console.log('\n📈 Inserting into new default database (analytics)...');
|
||||
const analyticsResult = await client.batchUpsert('test_metrics', analyticsData, 'name');
|
||||
console.log('Analytics database result:', analyticsResult);
|
||||
|
||||
// Test 5: Explicitly specify database (override default)
|
||||
const tradingData: TestDocument[] = [
|
||||
{ name: 'NYSE', value: 1 },
|
||||
{ name: 'NASDAQ', value: 2 },
|
||||
{ name: 'LSE', value: 3 },
|
||||
];
|
||||
|
||||
console.log('\n🏦 Inserting into specific database (trading_documents)...');
|
||||
const tradingResult = await client.batchUpsert('test_exchanges', tradingData, 'name', {
|
||||
database: 'trading_documents',
|
||||
});
|
||||
console.log('Trading documents database result:', tradingResult);
|
||||
|
||||
// Test 6: Use convenience methods
|
||||
console.log('\n🚀 Testing convenience methods...');
|
||||
|
||||
// Stock convenience method
|
||||
const stockConvenienceResult = await client.batchUpsertStock(
|
||||
'test_prices',
|
||||
[{ name: 'TSLA', value: 800.25 }],
|
||||
'name'
|
||||
);
|
||||
console.log('Stock convenience method result:', stockConvenienceResult);
|
||||
|
||||
// Analytics convenience method
|
||||
const analyticsConvenienceResult = await client.batchUpsertAnalytics(
|
||||
'test_performance',
|
||||
[{ name: 'sharpe_ratio', value: 1.25 }],
|
||||
'name'
|
||||
);
|
||||
console.log('Analytics convenience method result:', analyticsConvenienceResult);
|
||||
|
||||
// Trading convenience method
|
||||
const tradingConvenienceResult = await client.batchUpsertTrading(
|
||||
'test_orders',
|
||||
[{ name: 'order_001', value: 100 }],
|
||||
'name'
|
||||
);
|
||||
console.log('Trading convenience method result:', tradingConvenienceResult);
|
||||
|
||||
// Test 7: Direct database access
|
||||
console.log('\n🎯 Testing direct database access...');
|
||||
const stockDb = client.getDatabase('stock');
|
||||
const analyticsDb = client.getDatabase('analytics');
|
||||
const tradingDb = client.getDatabase('trading_documents');
|
||||
|
||||
console.log('Available databases:', {
|
||||
stock: stockDb.databaseName,
|
||||
analytics: analyticsDb.databaseName,
|
||||
trading: tradingDb.databaseName,
|
||||
});
|
||||
|
||||
// Test 8: Collection operations with database override
|
||||
console.log('\n📋 Testing collection operations with database override...');
|
||||
|
||||
const stockCollection = client.getCollection('test_symbols', 'stock');
|
||||
const stockCount = await stockCollection.countDocuments();
|
||||
console.log(`Stock test_symbols count: ${stockCount}`);
|
||||
|
||||
const analyticsCollection = client.getCollection('test_metrics', 'analytics');
|
||||
const analyticsCount = await analyticsCollection.countDocuments();
|
||||
console.log(`Analytics test_metrics count: ${analyticsCount}`);
|
||||
|
||||
// Test 9: InsertOne with database override
|
||||
console.log('\n➕ Testing insertOne with database override...');
|
||||
const insertResult = await client.insertOne(
|
||||
'test_single',
|
||||
{ name: 'single_test', value: 999 },
|
||||
'stock'
|
||||
);
|
||||
console.log('InsertOne result:', insertResult);
|
||||
|
||||
console.log('\n🎉 All multi-database tests completed successfully!');
|
||||
} catch (error) {
|
||||
console.error('❌ Test failed:', error);
|
||||
} finally {
|
||||
console.log('\n🔌 Disconnecting from MongoDB...');
|
||||
await client.disconnect();
|
||||
console.log('✅ Disconnected successfully!');
|
||||
}
|
||||
}
|
||||
|
||||
// Run the test
|
||||
testMultiDatabaseSupport().catch(console.error);
|
||||
Loading…
Add table
Add a link
Reference in a new issue