wehub-resource-sync
38 KiB
38 KiB
Najlepšie postupy a optimalizácia
🎯 Čo tento lab zahŕňa
Tento záverečný lab zhrňuje najlepšie postupy, techniky optimalizácie a výrobných usmernení pre budovanie robustných, škálovateľných a bezpečných MCP serverov s integráciou databázy. Naučíte sa z reálnej praxe a priemyselných štandardov, aby vaša implementácia bola pripravená na produkciu.
Prehľad
Vybudovanie úspešného MCP servera nie je len o tom, aby kód fungoval. Tento lab pokrýva základné postupy, ktoré odlišujú implementácie dôkazu konceptu od výrobných systémov, ktoré vedia škálovať, spoľahlivo pracovať a zachovávať bezpečnostné štandardy.
Tieto najlepšie postupy sú odvodené z reálnych nasadení, spätnej väzby komunity a skúseností z podnikových implementácií.
Ciele učenia
Na konci tohto labu budete schopní:
- Použiť techniky optimalizácie výkonu pre MCP servery a databázy
- Implementovať komplexné bezpečnostné opatrenia
- Navrhnúť škálovateľné architektonické vzory pre produkčné prostredia
- Zaviesť monitorovanie, údržbu a prevádzkové postupy
- Optimalizovať náklady pri zachovaní výkonu a spoľahlivosti
- Prispieť do komunity a ekosystému MCP
🚀 Optimalizácia výkonu
Výkon databázy
Optimalizácia pripojovacích poolov
# Optimalizovaná konfigurácia poolu pripojení
POOL_CONFIG = {
# Konfigurácia veľkosti
"min_size": max(2, cpu_count()), # Aspoň 2, prispôsobuje sa CPU
"max_size": min(20, cpu_count() * 4), # Limit na rozumnú maximálnu hodnotu
# Konfigurácia časovania
"max_inactive_connection_lifetime": 300, # 5 minút
"command_timeout": 30, # 30 sekúnd
"max_queries": 50000, # Rotovať pripojenia
# Nastavenia PostgreSQL
"server_settings": {
"application_name": "mcp-server-prod",
"jit": "off", # Zakázať pre konzistenciu
"work_mem": "8MB", # Optimalizovať pre dotazy
"shared_preload_libraries": "pg_stat_statements",
"log_statement": "mod", # Logovať len zmeny
"log_min_duration_statement": "1s", # Logovať pomalé dotazy
}
}
Vzory optimalizácie dotazov
class QueryOptimizer:
"""Database query optimization utilities."""
def __init__(self):
self.query_cache = {}
self.slow_query_threshold = 1.0 # sekundy
async def execute_optimized_query(
self,
query: str,
params: tuple = None,
cache_key: str = None,
cache_ttl: int = 300
):
"""Execute query with optimization and caching."""
# Najskôr skontrolujte vyrovnávaciu pamäť
if cache_key and cache_key in self.query_cache:
cache_entry = self.query_cache[cache_key]
if time.time() - cache_entry['timestamp'] < cache_ttl:
return cache_entry['result']
# Vykonať s monitorovaním
start_time = time.time()
try:
async with db_provider.get_connection() as conn:
# Optimalizovať vykonávanie dotazu
await conn.execute("SET enable_seqscan = off") # Uprednostniť indexy
await conn.execute("SET work_mem = '16MB'") # Viac pamäte pre tento dotaz
result = await conn.fetch(query, *params if params else ())
duration = time.time() - start_time
# Zaznamenávať pomalé dotazy
if duration > self.slow_query_threshold:
logger.warning(f"Slow query detected: {duration:.2f}s", extra={
"query": query[:200],
"duration": duration,
"params_count": len(params) if params else 0
})
# Ukladať úspešné výsledky do vyrovnávacej pamäte
if cache_key and len(result) < 1000: # Neukladať veľké výsledky do vyrovnávacej pamäte
self.query_cache[cache_key] = {
'result': result,
'timestamp': time.time()
}
return result
except Exception as e:
logger.error(f"Query optimization failed: {e}")
raise
# Odporúčania indexov
RECOMMENDED_INDEXES = [
# Základné obchodné indexy
"CREATE INDEX CONCURRENTLY idx_orders_store_date ON retail.orders (store_id, order_date DESC);",
"CREATE INDEX CONCURRENTLY idx_order_items_product ON retail.order_items (product_id);",
"CREATE INDEX CONCURRENTLY idx_customers_store_email ON retail.customers (store_id, email);",
# Analytické indexy
"CREATE INDEX CONCURRENTLY idx_orders_date_amount ON retail.orders (order_date, total_amount);",
"CREATE INDEX CONCURRENTLY idx_products_category_price ON retail.products (category_id, unit_price);",
# Optimalizácia vyhľadávania vektorov
"CREATE INDEX CONCURRENTLY idx_embeddings_vector ON retail.product_description_embeddings USING ivfflat (description_embedding vector_cosine_ops) WITH (lists = 100);",
]
Výkon aplikácie
Najlepšie praktiky asynchrónneho programovania
import asyncio
from asyncio import Semaphore
from typing import List, Any
class AsyncOptimizer:
"""Async operation optimization patterns."""
def __init__(self, max_concurrent: int = 10):
self.semaphore = Semaphore(max_concurrent)
self.circuit_breaker = CircuitBreaker()
async def batch_process(
self,
items: List[Any],
process_func: callable,
batch_size: int = 100
):
"""Process items in optimized batches."""
async def process_batch(batch):
async with self.semaphore:
return await asyncio.gather(
*[process_func(item) for item in batch],
return_exceptions=True
)
# Spracovanie po dávkach, aby sa predišlo preťaženiu systému
results = []
for i in range(0, len(items), batch_size):
batch = items[i:i + batch_size]
batch_results = await process_batch(batch)
results.extend(batch_results)
# Malé oneskorenie medzi dávkami, aby sa zabránilo vyčerpaniu zdrojov
if i + batch_size < len(items):
await asyncio.sleep(0.1)
return results
@circuit_breaker_decorator
async def resilient_operation(self, operation: callable, *args, **kwargs):
"""Execute operation with circuit breaker protection."""
return await operation(*args, **kwargs)
# Implementácia ističa obvodu
class CircuitBreaker:
"""Circuit breaker for external service calls."""
def __init__(self, failure_threshold: int = 5, recovery_timeout: int = 60):
self.failure_threshold = failure_threshold
self.recovery_timeout = recovery_timeout
self.failure_count = 0
self.last_failure_time = None
self.state = "CLOSED" # ZATVORENÉ, OTVORENÉ, POL-OTVORENÉ
async def call(self, func, *args, **kwargs):
"""Execute function with circuit breaker protection."""
if self.state == "OPEN":
if time.time() - self.last_failure_time > self.recovery_timeout:
self.state = "HALF_OPEN"
else:
raise Exception("Circuit breaker is OPEN")
try:
result = await func(*args, **kwargs)
# Resetovať pri úspechu
if self.state == "HALF_OPEN":
self.state = "CLOSED"
self.failure_count = 0
return result
except Exception as e:
self.failure_count += 1
self.last_failure_time = time.time()
if self.failure_count >= self.failure_threshold:
self.state = "OPEN"
raise
Stratégie cachovania
import redis
import pickle
from typing import Union, Optional
class SmartCache:
"""Multi-level caching system."""
def __init__(self, redis_url: Optional[str] = None):
self.memory_cache = {}
self.redis_client = redis.Redis.from_url(redis_url) if redis_url else None
self.max_memory_items = 1000
async def get(self, key: str) -> Optional[Any]:
"""Get from cache with fallback levels."""
# Úroveň 1: Pamäťová cache
if key in self.memory_cache:
return self.memory_cache[key]['value']
# Úroveň 2: Redis cache
if self.redis_client:
try:
cached_data = self.redis_client.get(key)
if cached_data:
value = pickle.loads(cached_data)
# Propagovať do pamäťovej cache
self._set_memory_cache(key, value)
return value
except Exception as e:
logger.warning(f"Redis cache error: {e}")
return None
async def set(
self,
key: str,
value: Any,
ttl: int = 300,
cache_level: str = "both"
):
"""Set cache value at specified levels."""
if cache_level in ["memory", "both"]:
self._set_memory_cache(key, value, ttl)
if cache_level in ["redis", "both"] and self.redis_client:
try:
self.redis_client.setex(
key,
ttl,
pickle.dumps(value)
)
except Exception as e:
logger.warning(f"Redis set error: {e}")
def _set_memory_cache(self, key: str, value: Any, ttl: int = 300):
"""Set value in memory cache with LRU eviction."""
# Implementovať LRU vyraďovanie
if len(self.memory_cache) >= self.max_memory_items:
oldest_key = min(
self.memory_cache.keys(),
key=lambda k: self.memory_cache[k]['timestamp']
)
del self.memory_cache[oldest_key]
self.memory_cache[key] = {
'value': value,
'timestamp': time.time(),
'ttl': ttl
}
# Generovanie cache kľúča
def generate_cache_key(query: str, user_context: str, params: dict = None) -> str:
"""Generate consistent cache keys."""
key_components = [
query.strip().lower(),
user_context,
json.dumps(params, sort_keys=True) if params else ""
]
key_string = "|".join(key_components)
return hashlib.sha256(key_string.encode()).hexdigest()
🔒 Bezpečnostné spevnenie
Autentifikácia a autorizácia
from azure.identity import DefaultAzureCredential, ClientSecretCredential
from azure.keyvault.secrets import SecretClient
import jwt
from typing import Dict, List
class SecurityManager:
"""Comprehensive security management."""
def __init__(self):
self.key_vault_client = self._setup_key_vault()
self.token_blacklist = set()
def _setup_key_vault(self) -> SecretClient:
"""Initialize Azure Key Vault client."""
credential = DefaultAzureCredential()
vault_url = os.getenv("AZURE_KEY_VAULT_URL")
if vault_url:
return SecretClient(vault_url=vault_url, credential=credential)
return None
async def validate_request(self, request_headers: Dict[str, str]) -> Dict[str, Any]:
"""Comprehensive request validation."""
# Extrahovať a overiť autentifikáciu
auth_token = request_headers.get("authorization", "").replace("Bearer ", "")
if not auth_token:
raise AuthenticationError("Missing authentication token")
# Overiť token
user_context = await self._validate_token(auth_token)
# Skontrolovať obmedzenie rýchlosti
await self._check_rate_limit(user_context["user_id"])
# Overiť kontext RLS
rls_user_id = request_headers.get("x-rls-user-id")
if not self._validate_rls_access(user_context, rls_user_id):
raise AuthorizationError("Invalid RLS context for user")
return {
"user_id": user_context["user_id"],
"roles": user_context["roles"],
"rls_user_id": rls_user_id,
"permissions": user_context["permissions"]
}
async def _validate_token(self, token: str) -> Dict[str, Any]:
"""Validate JWT token."""
if token in self.token_blacklist:
raise AuthenticationError("Token has been revoked")
try:
# Získať verejný kľúč z Key Vault alebo cache
public_key = await self._get_public_key()
# Dekódovať a overiť token
payload = jwt.decode(
token,
public_key,
algorithms=["RS256"],
audience="mcp-server",
issuer="zava-auth"
)
return {
"user_id": payload["sub"],
"roles": payload.get("roles", []),
"permissions": payload.get("permissions", []),
"expires_at": payload["exp"]
}
except jwt.InvalidTokenError as e:
raise AuthenticationError(f"Invalid token: {e}")
def _validate_rls_access(self, user_context: Dict, rls_user_id: str) -> bool:
"""Validate RLS context access."""
# Super administrátori môžu pristupovať k akémukoľvek kontextu
if "super_admin" in user_context["roles"]:
return True
# Manažéri obchodov môžu pristupovať iba k svojmu vlastnému obchodu
if "store_manager" in user_context["roles"]:
allowed_stores = user_context.get("allowed_stores", [])
return rls_user_id in allowed_stores
# Regionálni manažéri môžu pristupovať k viacerým obchodom
if "regional_manager" in user_context["roles"]:
allowed_regions = user_context.get("allowed_regions", [])
return self._check_store_in_regions(rls_user_id, allowed_regions)
return False
# Overenie a sanitizácia vstupu
class InputValidator:
"""SQL injection prevention and input validation."""
@staticmethod
def validate_sql_query(query: str) -> bool:
"""Validate SQL query for safety."""
# Zakázané vzory
forbidden_patterns = [
r";\s*(DROP|DELETE|UPDATE|INSERT|ALTER|CREATE)\s+",
r"--.*",
r"/\*.*\*/",
r"xp_cmdshell",
r"sp_executesql",
r"EXEC\s*\(",
]
query_upper = query.upper()
for pattern in forbidden_patterns:
if re.search(pattern, query_upper, re.IGNORECASE):
logger.warning(f"Blocked potentially dangerous query: {pattern}")
return False
# Povoliť iba SELECT príkazy
if not query_upper.strip().startswith("SELECT"):
return False
return True
@staticmethod
def sanitize_table_name(table_name: str) -> str:
"""Sanitize table name input."""
# Povoliť iba alfanumerické znaky, podčiarkovník a bodku
if not re.match(r"^[a-zA-Z0-9_.]+$", table_name):
raise ValueError("Invalid table name format")
# Overiť voči povoleným tabuľkám
if table_name not in VALID_TABLES:
raise ValueError(f"Table {table_name} not allowed")
return table_name
Ochrana dát
from cryptography.fernet import Fernet
import hashlib
class DataProtection:
"""Data encryption and protection utilities."""
def __init__(self):
self.encryption_key = self._get_encryption_key()
self.cipher_suite = Fernet(self.encryption_key)
def _get_encryption_key(self) -> bytes:
"""Get encryption key from secure storage."""
# V produkcii získajte z Azure Key Vault
key_vault_secret = os.getenv("ENCRYPTION_KEY_SECRET_NAME")
if key_vault_secret and self.key_vault_client:
secret = self.key_vault_client.get_secret(key_vault_secret)
return secret.value.encode()
# Náhradné riešenie pre vývoj (nie pre produkciu!)
dev_key = os.getenv("DEV_ENCRYPTION_KEY")
if dev_key:
return dev_key.encode()
raise ValueError("No encryption key available")
def encrypt_sensitive_data(self, data: str) -> str:
"""Encrypt sensitive data."""
return self.cipher_suite.encrypt(data.encode()).decode()
def decrypt_sensitive_data(self, encrypted_data: str) -> str:
"""Decrypt sensitive data."""
return self.cipher_suite.decrypt(encrypted_data.encode()).decode()
@staticmethod
def hash_password(password: str, salt: str = None) -> tuple:
"""Hash password with salt."""
if not salt:
salt = os.urandom(32).hex()
password_hash = hashlib.pbkdf2_hmac(
'sha256',
password.encode(),
salt.encode(),
100000 # iterácie
).hex()
return password_hash, salt
@staticmethod
def mask_sensitive_logs(log_data: dict) -> dict:
"""Mask sensitive information in logs."""
sensitive_fields = [
'password', 'token', 'secret', 'key', 'authorization',
'x-api-key', 'client_secret', 'connection_string'
]
masked_data = log_data.copy()
for field in sensitive_fields:
if field in masked_data:
value = str(masked_data[field])
if len(value) > 4:
masked_data[field] = value[:2] + "*" * (len(value) - 4) + value[-2:]
else:
masked_data[field] = "***"
return masked_data
📊 Pokyny pre výrobnú prevádzku
Infraštruktúra ako kód
# azure-pipelines.yml
trigger:
branches:
include:
- main
- release/*
variables:
- group: mcp-server-secrets
- name: imageRepository
value: 'zava-mcp-server'
- name: containerRegistry
value: 'zavamcpregistry.azurecr.io'
stages:
- stage: Build
displayName: Build and Test
jobs:
- job: Build
displayName: Build
pool:
vmImage: ubuntu-latest
steps:
- task: UsePythonVersion@0
inputs:
versionSpec: '3.11'
displayName: 'Use Python 3.11'
- script: |
python -m pip install --upgrade pip
pip install -r requirements.lock.txt
pip install pytest pytest-cov
displayName: 'Install dependencies'
- script: |
pytest tests/ --cov=mcp_server --cov-report=xml
displayName: 'Run tests with coverage'
- task: PublishCodeCoverageResults@1
inputs:
codeCoverageTool: Cobertura
summaryFileLocation: 'coverage.xml'
- task: Docker@2
displayName: Build Docker image
inputs:
command: build
repository: $(imageRepository)
dockerfile: Dockerfile
tags: |
$(Build.BuildId)
latest
- stage: Deploy
displayName: Deploy to Production
dependsOn: Build
condition: and(succeeded(), eq(variables['Build.SourceBranch'], 'refs/heads/main'))
jobs:
- deployment: DeployProduction
displayName: Deploy to Production
environment: 'production'
pool:
vmImage: ubuntu-latest
strategy:
runOnce:
deploy:
steps:
- task: AzureContainerApps@1
inputs:
azureSubscription: $(azureServiceConnection)
containerAppName: 'zava-mcp-server'
resourceGroup: '$(resourceGroupName)'
imageToDeploy: '$(containerRegistry)/$(imageRepository):$(Build.BuildId)'
Optimalizácia kontajnerov
# Multi-stage Dockerfile for production
FROM python:3.11-slim as builder
# Install build dependencies
RUN apt-get update && apt-get install -y \
gcc \
g++ \
&& rm -rf /var/lib/apt/lists/*
# Create virtual environment
RUN python -m venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# Copy requirements and install Python dependencies
COPY requirements.lock.txt .
RUN pip install --no-cache-dir --upgrade pip && \
pip install --no-cache-dir -r requirements.lock.txt
# Production stage
FROM python:3.11-slim as production
# Create non-root user
RUN groupadd -r mcpserver && useradd -r -g mcpserver mcpserver
# Copy virtual environment from builder
COPY --from=builder /opt/venv /opt/venv
ENV PATH="/opt/venv/bin:$PATH"
# Set working directory
WORKDIR /app
# Copy application code
COPY mcp_server/ ./mcp_server/
COPY --chown=mcpserver:mcpserver . .
# Set security configurations
RUN chmod -R 755 /app && \
chown -R mcpserver:mcpserver /app
# Switch to non-root user
USER mcpserver
# Health check
HEALTHCHECK --interval=30s --timeout=10s --start-period=5s --retries=3 \
CMD curl -f http://localhost:8000/health || exit 1
# Expose port
EXPOSE 8000
# Start application
CMD ["python", "-m", "mcp_server.sales_analysis"]
Konfigurácia prostredia
# Správa konfigurácie produkcie
class ProductionConfig:
"""Production-specific configuration."""
def __init__(self):
self.validate_production_requirements()
self.setup_logging()
self.configure_security()
def validate_production_requirements(self):
"""Validate all required production settings."""
required_settings = [
"AZURE_CLIENT_ID",
"AZURE_CLIENT_SECRET",
"AZURE_TENANT_ID",
"PROJECT_ENDPOINT",
"AZURE_OPENAI_ENDPOINT",
"POSTGRES_HOST",
"POSTGRES_PASSWORD",
"APPLICATIONINSIGHTS_CONNECTION_STRING"
]
missing_settings = [
setting for setting in required_settings
if not os.getenv(setting)
]
if missing_settings:
raise EnvironmentError(
f"Missing required production settings: {missing_settings}"
)
def setup_logging(self):
"""Configure production logging."""
logging.basicConfig(
level=logging.INFO,
format='%(asctime)s - %(name)s - %(levelname)s - %(message)s',
handlers=[
logging.StreamHandler(sys.stdout),
logging.handlers.RotatingFileHandler(
'/var/log/mcp-server.log',
maxBytes=50*1024*1024, # 50MB
backupCount=5
)
]
)
# Nastaviť logovacie nástroje tretích strán na UPOZORNENIE
logging.getLogger('azure').setLevel(logging.WARNING)
logging.getLogger('urllib3').setLevel(logging.WARNING)
def configure_security(self):
"""Configure production security settings."""
# Vypnúť režim ladenia
os.environ['DEBUG'] = 'False'
# Nastaviť zabezpečené hlavičky
os.environ['SECURE_SSL_REDIRECT'] = 'True'
os.environ['SECURE_HSTS_SECONDS'] = '31536000'
os.environ['SECURE_CONTENT_TYPE_NOSNIFF'] = 'True'
os.environ['SECURE_BROWSER_XSS_FILTER'] = 'True'
💰 Optimalizácia nákladov
Riadenie zdrojov
class CostOptimizer:
"""Cost optimization strategies."""
def __init__(self):
self.metrics_collector = MetricsCollector()
self.auto_scaler = AutoScaler()
async def optimize_database_connections(self):
"""Dynamically adjust connection pool based on load."""
current_load = await self.metrics_collector.get_current_load()
if current_load < 0.3: # Nízke zaťaženie
target_pool_size = max(2, int(current_load * 10))
elif current_load < 0.7: # Stredné zaťaženie
target_pool_size = max(5, int(current_load * 15))
else: # Vysoké zaťaženie
target_pool_size = min(20, int(current_load * 25))
await db_provider.adjust_pool_size(target_pool_size)
logger.info(f"Adjusted pool size to {target_pool_size} for load {current_load}")
async def implement_smart_caching(self):
"""Implement intelligent caching to reduce compute costs."""
# Ukladať nákladné operácie do vyrovnávacej pamäte
expensive_queries = await self.identify_expensive_queries()
for query in expensive_queries:
cache_key = self.generate_cache_key(query)
ttl = self.calculate_optimal_ttl(query)
await smart_cache.set(cache_key, None, ttl=ttl)
def calculate_azure_costs(self) -> Dict[str, float]:
"""Calculate estimated Azure resource costs."""
return {
"container_apps": self.estimate_container_costs(),
"postgresql": self.estimate_database_costs(),
"openai": self.estimate_ai_costs(),
"application_insights": self.estimate_monitoring_costs(),
"storage": self.estimate_storage_costs()
}
# Konfigurácia autoškálovania
class AutoScaler:
"""Automatic scaling based on metrics."""
async def scale_decision(self) -> str:
"""Determine scaling action based on metrics."""
metrics = await self.collect_scaling_metrics()
# Škálovanie na základe CPU
if metrics['cpu_usage'] > 80:
return "scale_up"
elif metrics['cpu_usage'] < 20 and metrics['instance_count'] > 1:
return "scale_down"
# Škálovanie na základe pamäte
if metrics['memory_usage'] > 85:
return "scale_up"
# Škálovanie frontu požiadaviek
if metrics['queue_length'] > 100:
return "scale_up"
elif metrics['queue_length'] < 10 and metrics['instance_count'] > 1:
return "scale_down"
return "no_action"
🔧 Údržba a prevádzka
Monitorovanie zdravia
class OperationalHealth:
"""Comprehensive operational health monitoring."""
def __init__(self):
self.alert_manager = AlertManager()
self.health_checks = {}
async def comprehensive_health_check(self) -> Dict[str, Any]:
"""Perform comprehensive system health check."""
health_report = {
"timestamp": datetime.utcnow().isoformat(),
"overall_status": "healthy",
"components": {}
}
# Stav databázy
db_health = await self.check_database_health()
health_report["components"]["database"] = db_health
# Stav externých služieb
ai_health = await self.check_ai_service_health()
health_report["components"]["ai_service"] = ai_health
# Systémové zdroje
system_health = await self.check_system_resources()
health_report["components"]["system"] = system_health
# Metričky aplikácie
app_health = await self.check_application_health()
health_report["components"]["application"] = app_health
# Určiť celkový stav
failed_components = [
name for name, status in health_report["components"].items()
if status.get("status") != "healthy"
]
if failed_components:
health_report["overall_status"] = "unhealthy"
health_report["failed_components"] = failed_components
# Spustiť upozornenia
await self.alert_manager.send_alert(
severity="high",
message=f"Health check failed for: {failed_components}",
details=health_report
)
return health_report
async def check_database_health(self) -> Dict[str, Any]:
"""Check database connectivity and performance."""
try:
start_time = time.time()
async with db_provider.get_connection() as conn:
# Základná konektivita
await conn.fetchval("SELECT 1")
# Skontrolovať pomalé dotazy
slow_queries = await conn.fetch("""
SELECT query, mean_exec_time, calls
FROM pg_stat_statements
WHERE mean_exec_time > 1000
ORDER BY mean_exec_time DESC
LIMIT 5
""")
# Skontrolovať počet pripojení
connection_count = await conn.fetchval("""
SELECT count(*) FROM pg_stat_activity
WHERE state = 'active'
""")
response_time = time.time() - start_time
return {
"status": "healthy",
"response_time_ms": response_time * 1000,
"active_connections": connection_count,
"slow_queries_count": len(slow_queries),
"pool_size": db_provider.connection_pool.get_size()
}
except Exception as e:
return {
"status": "unhealthy",
"error": str(e),
"last_check": datetime.utcnow().isoformat()
}
# Automatizované zálohovanie a obnova
class BackupManager:
"""Database backup and recovery management."""
async def create_backup(self, backup_type: str = "full") -> str:
"""Create database backup."""
timestamp = datetime.utcnow().strftime("%Y%m%d_%H%M%S")
backup_name = f"zava_backup_{backup_type}_{timestamp}"
if backup_type == "full":
await self.create_full_backup(backup_name)
elif backup_type == "incremental":
await self.create_incremental_backup(backup_name)
# Nahrávanie do Azure Blob Storage
await self.upload_backup_to_azure(backup_name)
return backup_name
async def schedule_automated_backups(self):
"""Schedule regular automated backups."""
# Denná úplná záloha o 2:00 UTC
schedule.every().day.at("02:00").do(
lambda: asyncio.create_task(self.create_backup("full"))
)
# Hodinové inkrementálne zálohy
schedule.every().hour.do(
lambda: asyncio.create_task(self.create_backup("incremental"))
)
🌍 Príspevky komunity
Najlepšie postupy open source
# Contributing to MCP Database Integration
## Development Guidelines
### Code Quality Standards
- Follow PEP 8 for Python code style
- Maintain test coverage above 90%
- Use type hints throughout the codebase
- Write comprehensive docstrings
### Testing Requirements
- Unit tests for all new functionality
- Integration tests for database operations
- Performance benchmarks for critical paths
- Security tests for authentication/authorization
### Documentation Standards
- Update README.md for any new features
- Add inline code documentation
- Create examples for new tools or patterns
- Maintain API documentation
## Security Considerations
### Reporting Security Issues
- Report security vulnerabilities privately
- Use encrypted communication channels
- Provide detailed reproduction steps
- Include potential impact assessment
### Security Review Process
- All PRs undergo security review
- Static analysis tools required to pass
- Dependency vulnerability scanning
- Manual security testing for critical changes
Zapojenie komunity
class CommunityContributor:
"""Tools for community engagement and contribution."""
@staticmethod
def generate_contribution_guide():
"""Generate personalized contribution guide."""
return {
"getting_started": {
"setup": "Follow setup guide in Lab 03",
"first_contribution": "Start with documentation improvements",
"testing": "Run full test suite before submitting PR"
},
"contribution_areas": {
"documentation": "Improve learning labs and examples",
"testing": "Add test cases and improve coverage",
"features": "Implement new MCP tools and capabilities",
"performance": "Optimize queries and caching",
"security": "Enhance security measures and validation"
},
"community_resources": {
"discord": "https://discord.com/invite/ByRwuEEgH4",
"discussions": "GitHub Discussions for Q&A",
"issues": "GitHub Issues for bug reports",
"examples": "Share your implementation examples"
}
}
@staticmethod
def validate_contribution(pr_data: Dict) -> Dict[str, bool]:
"""Validate contribution meets standards."""
return {
"has_tests": "test" in pr_data.get("files_changed", []),
"has_documentation": "README" in str(pr_data.get("files_changed", [])),
"follows_conventions": True, # Skutočné kontroly by sa implementovali
"security_reviewed": pr_data.get("security_review", False),
"performance_tested": pr_data.get("benchmark_results", False)
}
🎯 Kľúčové zhrnutia
Po dokončení tohto komplexného učebného plánu by ste mali zvládnuť:
✅ Optimalizácia výkonu: ladenie databázy, asynchrónne vzory a strategie cachovania
✅ Bezpečnostné spevnenie: autentifikácia, autorizácia a ochrana dát
✅ Výrobná prevádzka: infraštruktúra ako kód a optimalizácia kontajnerov
✅ Riadenie nákladov: optimalizácia zdrojov a inteligentné škálovanie
✅ Prevádzková excelentnosť: monitorovanie, údržba a automatizácia
✅ Zapojenie komunity: príspevky do ekosystému MCP
🏆 Certifikácia a ďalšie kroky
Praktické hodnotenie
Dokončite tento záverečný projekt, aby ste preukázali svoje majstrovstvo:
Vybudujte produkčne pripravený MCP server, ktorý obsahuje:
- Analýzu maloobchodných dát s multi-tenant RLS
- Semantické vyhľadávanie s Azure OpenAI
- Komplexnú bezpečnostnú implementáciu
- Výrobnú prevádzku na Azure
- Nastavenie monitorovania a upozornení
- Dokumentáciu a testovanie
Pokročilé učebné cesty
Pokračujte vo svojej MCP ceste s:
- Architektonické vzory MCP: Pokročilé serverové architektúry
- Integrácia multimodelov: Kombinovanie rôznych AI modelov
- Podnikové škálovanie: Veľkosériové nasadenie MCP
- Vývoj vlastných nástrojov: Budovanie špecializovaných nástrojov MCP
- Ekosystém MCP: Príspevky do širšej komunity
Uznanie komunity
Podeľte sa o svoj úspech:
- GitHub portfólio: Predveďte svoju implementáciu
- Príspevky komunity: Posielajte vylepšenia alebo príklady
- Možnosti vystúpení: Prednášajte na stretnutiach alebo konferenciách
- Mentoring: Pomáhajte iným vývojárom učiť sa MCP
📚 Dodatočné zdroje
Pokročilé témy
- Ladenie výkonu PostgreSQL - Optimalizácia databázy
- Najlepšie praktiky Azure Container Apps - Výrobná prevádzka
- Najlepšie praktiky python.async - Asynchrónne programovanie
Bezpečnostné zdroje
- OWASP Top 10 - bezpečnostné slabiny
- Najlepšie praktiky zabezpečenia Azure - zabezpečenie cloudu
- Bezpečnostné usmernenia Python - bezpečné kódovanie
Komunita
- MCP Community Discord - živé diskusie
- GitHub diskusie - otázky a zdieľanie
- Stack Overflow - technické otázky
🎉 Gratulujeme! Úspešne ste zvládli komplexný učebný plán pre integráciu MCP databázy. Teraz máte vedomosti a schopnosti budovať produkčne pripravené MCP servery, ktoré spájajú AI asistenta s reálnymi dátovými systémami.
Pripravení prispieť? Pridajte sa do našej komunity a pomáhajte iným učiť sa MCP zdieľaním skúseností, vylepšovaním kódu alebo tvorbou ďalších učebných zdrojov.
Ďalšie: Tooling
Vyhlásenie o zodpovednosti: Tento dokument bol preložený pomocou AI prekladateľskej služby Co-op Translator. Hoci sa snažíme o presnosť, vezmite prosím na vedomie, že automatické preklady môžu obsahovať chyby alebo nepresnosti. Pôvodný dokument v jeho natívnom jazyku by mal byť považovaný za autoritatívny zdroj. Pre kritické informácie sa odporúča profesionálny ľudský preklad. Nie sme zodpovední za žiadne nedorozumenia alebo nesprávne interpretácie vyplývajúce z použitia tohto prekladu.