admin/Nexus
1
0
Fork 0
Code Issues Pull Requests Actions 91 Packages Projects Releases Wiki Activity

Nexus v6.0.0: Initialize project structure with Clean Architecture

Browse Source 
- Project brand: Nexus (configurable via settings.SYSTEM_NAME)
- Clean Architecture 4-layer directory structure:
  server/domain/models/ — SQLAlchemy models
  server/domain/repositories/ — Abstract interfaces (Protocol)
  server/infrastructure/database/ — Async SQLAlchemy session + crypto
  server/infrastructure/redis/ — Decoupled Redis client
  server/infrastructure/ssh/ — Unified SSH connection pool
  server/application/services/ — Business logic (TODO)
  server/api/ — FastAPI routes (TODO)
- Config: Python 3.12+, PHP 8.2+, MySQL 8.4 LTS, Redis 7.0+
- Async SQLAlchemy (aiomysql) replacing sync pymysql
- Test framework: pytest + async fixtures + Playwright (conftest.py)
- MCP server address: 47.254.123.106

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>
This commit is contained in:
Your Name
2026-05-20 14:19:09 +08:00
commit b2617b4e31
23 changed files with 806 additions and 0 deletions
Download Patch File Download Diff File Expand all files Collapse all files
Nexus/.env.example
+33
View File
@@ -0,0 +1,33 @@
# Nexus — Environment Configuration
# Copy to .env and fill in your values
# Brand (configurable in MySQL settings table)
NEXUS_SYSTEM_NAME=Nexus
NEXUS_SYSTEM_TITLE=Nexus — 服务器运维管理平台
# Server
NEXUS_HOST=0.0.0.0
NEXUS_PORT=8600
# Database (MySQL 8.4+)
NEXUS_DATABASE_URL=mysql+aiomysql://root:password@127.0.0.1:3306/nexus
NEXUS_DB_POOL_SIZE=100
NEXUS_DB_MAX_OVERFLOW=100
# Security — CHANGE THESE IN PRODUCTION!
NEXUS_SECRET_KEY=change-me-use-openssl-rand-hex-32
NEXUS_API_KEY=
NEXUS_ENCRYPTION_KEY=
# Redis
NEXUS_REDIS_URL=redis://127.0.0.1:6379/0
# SSH
NEXUS_SSH_STRICT_HOST_CHECKING=true
# Health Check
NEXUS_HEALTH_CHECK_INTERVAL=60
# Telegram Alerts (optional)
NEXUS_TELEGRAM_BOT_TOKEN=
NEXUS_TELEGRAM_CHAT_ID=
Nexus/.gitignore
+49
View File
@@ -0,0 +1,49 @@
# Nexus — Python
__pycache__/
*.py[cod]
*$py.class
*.egg-info/
dist/
build/
.eggs/
# Environment
.env
.venv/
venv/
ENV/
# IDE
.vscode/
.idea/
*.swp
*.swo
*~
# OS
.DS_Store
Thumbs.db
# Logs
*.log
nexus.log
# Test
.pytest_cache/
htmlcov/
.coverage
coverage.xml
# Node (for Tailwind/Alpine.js build)
node_modules/
web/css/tailwind-output.css
# Uploads
web/uploads/
# Data (sensitive)
web/data/config.php
web/data/*.db
# Deploy
deploy/*.key
Nexus/.mcp.json
+14
View File
@@ -0,0 +1,14 @@
{
"mcpServers": {
"multisync": {
"command": "ssh",
"args": [
"-i",
"/c/Users/uzuma/.ssh/id_rsa_mcp",
"root@47.254.123.106",
"python3",
"/www/wwwroot/nexus/mcp/nexus_server.py"
]
}
}
}
Nexus/requirements.txt
+23
View File
@@ -0,0 +1,23 @@
# Nexus — Server Management Platform
# Python Backend Dependencies
fastapi==0.115.6
uvicorn[standard]==0.34.0
sqlalchemy[asyncio]==2.0.36
aiomysql==0.2.0
alembic==1.14.0
pydantic==2.10.3
pydantic-settings==2.7.0
python-multipart==0.0.19
jinja2==3.1.5
aiofiles==24.1.0
httpx==0.28.1
apscheduler==3.11.0
websockets==14.1
paramiko==3.5.0
python-jose[cryptography]==3.3.0
passlib[bcrypt]==1.7.4
bcrypt==4.2.1
pyyaml==6.0.2
rich==13.9.4
redis[hiredis]==5.2.1
Nexus/server/__init__.py
+5
View File
@@ -0,0 +1,5 @@
"""Nexus — Server Management Platform
Configurable brand name via settings.SYSTEM_NAME (default: Nexus)
"""
__version__ = "6.0.0"
Nexus/server/api/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Nexus — API Layer (FastAPI routes + dependency injection)"""
Nexus/server/application/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Nexus — Application Layer (Business logic / Use cases)"""
Nexus/server/application/services/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Nexus — Application Services (Business logic)"""
Nexus/server/config.py
+44
View File
@@ -0,0 +1,44 @@
"""Nexus Configuration — Pydantic Settings with MySQL overrides"""
from pydantic_settings import BaseSettings
from typing import Optional
class Settings(BaseSettings):
"""Base settings loaded from .env file, overrideable via MySQL settings table"""
# Brand (configurable in system settings)
SYSTEM_NAME: str = "Nexus"
SYSTEM_TITLE: str = "Nexus — 服务器运维管理平台"
# Server
HOST: str = "0.0.0.0"
PORT: int = 8600
# Database
DATABASE_URL: str = "mysql+pymysql://root:password@127.0.0.1:3306/nexus"
DB_POOL_SIZE: Optional[int] = 100
DB_MAX_OVERFLOW: Optional[int] = 100
# Security
SECRET_KEY: str = "" # Must be set in .env or settings table; empty = startup error
API_KEY: str = ""
ENCRYPTION_KEY: str = ""
# Redis
REDIS_URL: str = "redis://127.0.0.1:6379/0"
# SSH
SSH_STRICT_HOST_CHECKING: str = "true"
# Health Check
HEALTH_CHECK_INTERVAL: int = 60
# Telegram
TELEGRAM_BOT_TOKEN: str = ""
TELEGRAM_CHAT_ID: str = ""
model_config = {"env_prefix": "NEXUS_", "env_file": ".env"}
settings = Settings()
Nexus/server/domain/__init__.py
+3
View File
@@ -0,0 +1,3 @@
"""Nexus — Domain Layer
Contains SQLAlchemy models and repository abstract interfaces.
"""
Nexus/server/domain/models/__init__.py
+221
View File
@@ -0,0 +1,221 @@
"""Nexus — Domain Models
SQLAlchemy ORM models for all database tables.
"""
import datetime
import base64
import hashlib
from sqlalchemy import (
create_engine, Column, Integer, String, Boolean,
DateTime, Text, ForeignKey, Enum, Index
)
from sqlalchemy.orm import declarative_base, relationship
from cryptography.fernet import Fernet
Base = declarative_base()
class Server(Base):
"""Sub-server configuration"""
__tablename__ = "servers"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(100), unique=True, nullable=False, comment="服务器名称")
domain = Column(String(255), nullable=False, comment="域名或IP")
port = Column(Integer, default=22, comment="SSH端口")
username = Column(String(100), default="root", comment="SSH用户名")
auth_method = Column(String(20), default="key", comment="认证方式: key/password")
password = Column(String(255), nullable=True, comment="SSH密码(加密存储)")
ssh_key_path = Column(String(500), nullable=True, comment="SSH私钥路径")
ssh_key_private = Column(String(500), nullable=True, comment="加密后的私钥内容")
ssh_key_public = Column(String(500), nullable=True, comment="公钥内容")
ssh_key_configured = Column(Boolean, default=False, comment="是否已配置SSH Key")
agent_port = Column(Integer, default=8601, comment="Agent API端口")
agent_api_key = Column(String(255), nullable=True, comment="Agent API密钥")
description = Column(Text, nullable=True, comment="备注说明")
target_path = Column(String(500), nullable=True, comment="推送目标路径")
category = Column(String(100), nullable=True, comment="服务器分类")
# Status (from Agent heartbeat)
is_online = Column(Boolean, default=False, comment="是否在线")
last_heartbeat = Column(DateTime, nullable=True, comment="最后心跳时间")
system_info = Column(Text, nullable=True, comment="系统信息JSON(Agent上报)")
agent_version = Column(String(20), nullable=True, comment="Agent版本")
# Metadata
created_at = Column(DateTime, default=datetime.datetime.utcnow)
updated_at = Column(DateTime, default=datetime.datetime.utcnow, onupdate=datetime.datetime.utcnow)
sync_logs = relationship("SyncLog", back_populates="server", cascade="all, delete-orphan")
__table_args__ = (
Index('idx_servers_is_online', 'is_online'),
Index('idx_servers_category', 'category'),
)
class SyncLog(Base):
"""Push operation log"""
__tablename__ = "sync_logs"
id = Column(Integer, primary_key=True, autoincrement=True)
server_id = Column(Integer, ForeignKey("servers.id"), nullable=False)
source_path = Column(String(500), nullable=False, comment="源目录")
target_path = Column(String(500), nullable=False, comment="目标目录")
trigger_type = Column(String(20), comment="触发类型: manual/schedule/batch")
operator = Column(String(100), nullable=True, comment="操作人用户名")
status = Column(String(20), comment="pending/running/success/failed")
sync_mode = Column(String(20), default="incremental", comment="同步模式: incremental/full/overwrite/checksum")
files_total = Column(Integer, default=0)
files_transferred = Column(Integer, default=0)
files_skipped = Column(Integer, default=0)
bytes_transferred = Column(Integer, default=0)
duration_seconds = Column(Integer, default=0)
diff_summary = Column(Text, nullable=True, comment="rsync输出摘要")
error_message = Column(Text, nullable=True)
started_at = Column(DateTime, default=datetime.datetime.utcnow)
finished_at = Column(DateTime, nullable=True)
server = relationship("Server", back_populates="sync_logs")
__table_args__ = (
Index('idx_sync_logs_srv_start', 'server_id', 'started_at'),
)
class Admin(Base):
"""Admin user for web login"""
__tablename__ = "admins"
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(100), unique=True, nullable=False)
password_hash = Column(String(255), nullable=False)
email = Column(String(255), nullable=True)
totp_secret = Column(String(64), nullable=True)
totp_enabled = Column(Boolean, default=False)
is_active = Column(Boolean, default=True)
created_at = Column(DateTime, default=datetime.datetime.utcnow)
last_login = Column(DateTime, nullable=True)
class LoginAttempt(Base):
"""Login attempt log for brute-force protection"""
__tablename__ = "login_attempts"
id = Column(Integer, primary_key=True, autoincrement=True)
username = Column(String(100), nullable=False)
ip_address = Column(String(45), nullable=False)
attempted_at = Column(DateTime, default=datetime.datetime.utcnow)
success = Column(Boolean, default=False)
class Setting(Base):
"""Key-value system settings (including brand config)"""
__tablename__ = "settings"
key = Column(String(100), primary_key=True)
value = Column(Text, nullable=True)
updated_at = Column(DateTime, default=datetime.datetime.utcnow)
class PasswordPreset(Base):
"""SSH password preset — encrypted storage"""
__tablename__ = "password_presets"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(100), nullable=False, comment="预设名称")
encrypted_pw = Column(String(500), nullable=False, comment="加密后的密码")
created_at = Column(DateTime, default=datetime.datetime.utcnow)
class PushSchedule(Base):
"""Scheduled push — cron expression trigger"""
__tablename__ = "push_schedules"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(100), nullable=False, comment="调度名称")
source_path = Column(String(500), nullable=False)
server_ids = Column(Text, nullable=False, comment="JSON数组: 目标服务器ID列表")
cron_expr = Column(String(100), nullable=False, comment="cron表达式: 分 时 日 月 周")
enabled = Column(Boolean, default=True)
last_run_at = Column(DateTime, nullable=True)
created_at = Column(DateTime, default=datetime.datetime.utcnow)
class PushRetryJob(Base):
"""Failed push retry queue"""
__tablename__ = "push_retry_jobs"
id = Column(Integer, primary_key=True, autoincrement=True)
server_id = Column(Integer, ForeignKey("servers.id"), nullable=False)
server_name = Column(String(100), comment="冗余字段便于列表显示")
operator = Column(String(100), nullable=True, comment="操作人用户名")
source_path = Column(String(500), nullable=False)
target_path = Column(String(500), nullable=False)
status = Column(String(20), default="pending")
retry_count = Column(Integer, default=0)
max_retries = Column(Integer, default=100)
next_retry_at = Column(DateTime)
last_error = Column(Text)
created_at = Column(DateTime, default=datetime.datetime.utcnow)
updated_at = Column(DateTime, default=datetime.datetime.utcnow)
class AuditLog(Base):
"""Audit log — record all key operations"""
__tablename__ = "audit_logs"
id = Column(Integer, primary_key=True, autoincrement=True)
admin_username = Column(String(100), nullable=True, comment="操作人用户名")
action = Column(String(50), nullable=False, comment="操作类型")
target_type = Column(String(50), nullable=True, comment="目标类型")
target_id = Column(Integer, nullable=True, comment="目标ID")
detail = Column(Text, nullable=True, comment="操作详情JSON")
ip_address = Column(String(45), nullable=True, comment="操作IP")
created_at = Column(DateTime, default=datetime.datetime.utcnow)
class Script(Base):
"""Script library — reusable shell scripts"""
__tablename__ = "scripts"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(100), nullable=False, comment="脚本名称")
category = Column(String(50), default="ops", comment="分类: ops/deploy/check/cleanup")
content = Column(Text, nullable=False, comment="Shell命令内容")
description = Column(Text, nullable=True, comment="说明")
created_by = Column(String(100), nullable=True, comment="创建人")
created_at = Column(DateTime, default=datetime.datetime.utcnow)
updated_at = Column(DateTime, default=datetime.datetime.utcnow)
class ScriptExecution(Base):
"""Script execution log — results per server"""
__tablename__ = "script_executions"
id = Column(Integer, primary_key=True, autoincrement=True)
script_id = Column(Integer, ForeignKey("scripts.id"), nullable=True, comment="关联脚本(手动输入时为null)")
command = Column(Text, nullable=False, comment="实际执行的命令")
server_ids = Column(Text, nullable=False, comment="JSON数组: 目标服务器ID列表")
status = Column(String(20), default="pending", comment="pending/running/completed/failed")
results = Column(Text, nullable=True, comment="JSON: 每台服务器执行结果{server_id: {stdout,stderr,exit_code}}")
operator = Column(String(100), nullable=True, comment="操作人用户名")
started_at = Column(DateTime, default=datetime.datetime.utcnow)
completed_at = Column(DateTime, nullable=True)
class DbCredential(Base):
"""Database credentials — encrypted storage for $DB_* variable substitution"""
__tablename__ = "db_credentials"
id = Column(Integer, primary_key=True, autoincrement=True)
name = Column(String(100), nullable=False, comment="凭据名称")
db_type = Column(String(20), default="mysql", comment="数据库类型: mysql/postgresql/etc")
host = Column(String(255), default="localhost", comment="数据库主机")
port = Column(Integer, default=3306, comment="数据库端口")
username = Column(String(100), nullable=False, comment="数据库用户名")
encrypted_password = Column(String(500), nullable=False, comment="加密后的密码")
database = Column(String(100), nullable=True, comment="数据库库名")
created_at = Column(DateTime, default=datetime.datetime.utcnow)
Nexus/server/domain/repositories/__init__.py
+55
View File
@@ -0,0 +1,55 @@
"""Nexus — Repository Abstract Interfaces
Protocol-based interfaces for dependency inversion.
Concrete implementations in infrastructure/database/.
"""
from typing import Protocol, Optional, List
from server.domain.models import Server, SyncLog, Script, ScriptExecution, DbCredential
class ServerRepository(Protocol):
"""Server data access abstract interface"""
async def get_by_id(self, id: int) -> Optional[Server]: ...
async def get_all(self) -> List[Server]: ...
async def get_by_category(self, category: str) -> List[Server]: ...
async def get_online(self) -> List[Server]: ...
async def create(self, server: Server) -> Server: ...
async def update(self, server: Server) -> Server: ...
async def delete(self, id: int) -> bool: ...
class SyncLogRepository(Protocol):
"""Sync log data access abstract interface"""
async def get_by_server_id(self, server_id: int, limit: int = 50) -> List[SyncLog]: ...
async def create(self, log: SyncLog) -> SyncLog: ...
async def update_status(self, id: int, status: str, **kwargs) -> SyncLog: ...
class ScriptRepository(Protocol):
"""Script library data access abstract interface"""
async def get_by_id(self, id: int) -> Optional[Script]: ...
async def get_all(self) -> List[Script]: ...
async def get_by_category(self, category: str) -> List[Script]: ...
async def create(self, script: Script) -> Script: ...
async def update(self, script: Script) -> Script: ...
async def delete(self, id: int) -> bool: ...
class ScriptExecutionRepository(Protocol):
"""Script execution data access abstract interface"""
async def get_by_id(self, id: int) -> Optional[ScriptExecution]: ...
async def create(self, execution: ScriptExecution) -> ScriptExecution: ...
async def update_status(self, id: int, status: str, results: str) -> ScriptExecution: ...
class DbCredentialRepository(Protocol):
"""Database credential data access abstract interface"""
async def get_by_id(self, id: int) -> Optional[DbCredential]: ...
async def get_all(self) -> List[DbCredential]: ...
async def create(self, credential: DbCredential) -> DbCredential: ...
async def delete(self, id: int) -> bool: ...
Nexus/server/infrastructure/__init__.py
+7
View File
@@ -0,0 +1,7 @@
"""Nexus — Infrastructure Layer
Concrete implementations of domain interfaces.
"""
from server.infrastructure.database.session import get_async_session, init_db
from server.infrastructure.redis.client import get_redis
from server.infrastructure.ssh.pool import get_ssh_pool
Nexus/server/infrastructure/database/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Nexus — Infrastructure Layer: Database concrete implementations"""
Nexus/server/infrastructure/database/crypto.py
+70
View File
@@ -0,0 +1,70 @@
"""Nexus — Encryption Utilities (Fernet + AES)
Shared crypto module — used by SSH pool, DB credentials, password presets.
"""
import base64
import hashlib
from cryptography.fernet import Fernet
from cryptography.hazmat.primitives.ciphers import Cipher, algorithms, modes
from server.config import settings
def _fernet() -> Fernet:
key = settings.ENCRYPTION_KEY
if not key:
raw = hashlib.sha256(settings.SECRET_KEY.encode()).digest()
key = base64.urlsafe_b64encode(raw).decode()
return Fernet(key.encode() if isinstance(key, str) else key)
_fernet_instance: Fernet | None = None
def get_fernet() -> Fernet:
global _fernet_instance
if _fernet_instance is None:
_fernet_instance = _fernet()
return _fernet_instance
def _derive_aes_key() -> bytes:
"""Unified key derivation: must match PHP _encryptPw()
PHP uses hash('sha256', API_KEY, true).
This uses API_KEY first, falls back to SECRET_KEY."""
key = settings.API_KEY or settings.SECRET_KEY
return hashlib.sha256(key.encode()).digest()
def encrypt_value(plaintext: str) -> str:
"""Encrypt a string using Fernet"""
if not plaintext:
return plaintext
return get_fernet().encrypt(plaintext.encode()).decode()
def decrypt_value(ciphertext: str) -> str:
"""Decrypt a string — supports both Fernet and PHP AES format"""
if not ciphertext:
return ciphertext
# PHP-compatible AES format
if ciphertext.startswith("aes:"):
try:
key = _derive_aes_key()
data = base64.b64decode(ciphertext[4:])
iv = data[:16]
enc = data[16:]
cipher = Cipher(algorithms.AES(key), modes.CBC(iv))
decryptor = cipher.decryptor()
plain = decryptor.update(enc) + decryptor.finalize()
pad_len = plain[-1]
if 1 <= pad_len <= 16:
plain = plain[:-pad_len]
return plain.decode()
except Exception:
return ciphertext
# Fernet format
try:
return get_fernet().decrypt(ciphertext.encode()).decode()
except Exception:
return ciphertext # Compatible with pre-encryption data
Nexus/server/infrastructure/database/session.py
+37
View File
@@ -0,0 +1,37 @@
"""Nexus — Database Session Management (Async SQLAlchemy)
"""
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from server.domain.models import Base
from server.config import settings
# Async engine with pool settings
engine = create_async_engine(
settings.DATABASE_URL.replace("mysql+pymysql", "mysql+aiomysql"),
pool_size=max(5, int(settings.DB_POOL_SIZE or 100)),
max_overflow=max(5, int(settings.DB_MAX_OVERFLOW or 100)),
pool_recycle=300,
pool_pre_ping=True,
)
AsyncSessionLocal = async_sessionmaker(
autocommit=False,
autoflush=False,
bind=engine,
class_=AsyncSession,
)
async def get_async_session():
"""Dependency injection: yield async DB session"""
session = AsyncSessionLocal()
try:
yield session
finally:
await session.close()
async def init_db():
"""Initialize database tables"""
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
Nexus/server/infrastructure/redis/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Nexus — Infrastructure Layer: Redis client"""
Nexus/server/infrastructure/redis/client.py
+26
View File
@@ -0,0 +1,26 @@
"""Nexus — Redis Client (decoupled from main.py)
"""
import redis.asyncio as aioredis
from server.config import settings
_redis_pool: aioredis.Redis | None = None
async def get_redis() -> aioredis.Redis | None:
"""Get Redis connection (optional — returns None if Redis unavailable)"""
global _redis_pool
if not settings.REDIS_URL:
return None
if _redis_pool is None:
try:
_redis_pool = aioredis.from_url(
settings.REDIS_URL,
decode_responses=True,
max_connections=50,
)
await _redis_pool.ping()
except Exception:
_redis_pool = None
return None
return _redis_pool
Nexus/server/infrastructure/ssh/__init__.py
+1
View File
@@ -0,0 +1 @@
"""Nexus — Infrastructure Layer: SSH connection pool"""
Nexus/server/infrastructure/ssh/pool.py
+95
View File
@@ -0,0 +1,95 @@
"""Nexus — SSH Connection Pool (Unified, replacing 4+ duplicated implementations)
Provides:
- SSHConfig dataclass for connection parameters
- ssh_connection() async context manager
- create_ssh_client() factory with SSH Key support
- Batch operations via async pool
"""
import asyncio
import contextlib
from dataclasses import dataclass
from typing import Optional
import paramiko
from server.domain.models import Server
from server.infrastructure.database.crypto import decrypt_value
@dataclass
class SSHConfig:
"""SSH connection configuration"""
host: str
port: int = 22
username: str = "root"
auth_method: str = "key" # key/password
password: Optional[str] = None
key_path: Optional[str] = None
key_content: Optional[str] = None # For in-memory key (from DB encrypted field)
strict_host_checking: str = "true"
@classmethod
def from_server(cls, server: Server) -> "SSHConfig":
"""Build SSHConfig from Server model (with decrypted password)"""
return cls(
host=server.domain,
port=server.port,
username=server.username,
auth_method=server.auth_method,
password=server.decrypted_password if server.auth_method == "password" else None,
key_path=server.ssh_key_path,
key_content=decrypt_value(server.ssh_key_private or "") if server.ssh_key_configured else None,
)
def create_ssh_client(config: SSHConfig) -> paramiko.SSHClient:
"""Create SSH client with key or password authentication"""
client = paramiko.SSHClient()
if config.strict_host_checking == "true":
client.set_missing_host_key_policy(paramiko.WarningPolicy())
else:
client.set_missing_host_key_policy(paramiko.AutoAddPolicy())
if config.auth_method == "key" and config.key_content:
# Use in-memory key from encrypted DB field
import io
key = paramiko.Ed25519Key.from_private_key(io.StringIO(config.key_content))
client.connect(config.host, port=config.port, username=config.username, pkey=key)
elif config.auth_method == "key" and config.key_path:
client.connect(config.host, port=config.port, username=config.username, key_filename=config.key_path)
elif config.auth_method == "password" and config.password:
client.connect(config.host, port=config.port, username=config.username, password=config.password)
else:
raise ValueError(f"No valid SSH credentials for {config.host}")
return client
@contextlib.asynccontextmanager
async def ssh_connection(config: SSHConfig):
"""Async context manager for SSH connections (creates and cleans up)"""
client = await asyncio.to_thread(create_ssh_client, config)
try:
yield client
finally:
client.close()
async def exec_command(config: SSHConfig, command: str, timeout: int = 30) -> dict:
"""Execute a command via SSH and return structured result"""
with await ssh_connection(config) as client:
stdin, stdout, stderr = await asyncio.to_thread(
client.exec_command, command, timeout=timeout
)
exit_code = stdout.channel.recv_exit_status()
out = await asyncio.to_thread(stdout.read)
err = await asyncio.to_thread(stderr.read)
return {
"status": "success" if exit_code == 0 else "failed",
"stdout": out.decode()[:10000],
"stderr": err.decode()[:10000],
"exit_code": exit_code,
}
Nexus/server/main.py
+57
View File
@@ -0,0 +1,57 @@
"""Nexus — FastAPI Application Entry Point
Clean Architecture: Presentation Layer (API routes + middleware + lifespan)
"""
import logging
from contextlib import asynccontextmanager
from fastapi import FastAPI
from fastapi.middleware.cors import CORSMiddleware
from server.config import settings
from server.infrastructure.database.session import init_db
logger = logging.getLogger("nexus")
@asynccontextmanager
async def lifespan(app: FastAPI):
"""Application lifecycle: startup → background tasks → shutdown"""
# Startup
if not settings.SECRET_KEY:
logger.error("SECRET_KEY is empty! Set it in .env or MySQL settings table.")
raise SystemExit("SECRET_KEY is required for Nexus to start.")
await init_db()
logger.info(f"{settings.SYSTEM_NAME} started — DB initialized")
# TODO: Start background tasks (retry_loop, scheduler_loop, monitor_loop, heartbeat flush)
yield
# Shutdown
logger.info(f"{settings.SYSTEM_NAME} shutting down")
app = FastAPI(
title=settings.SYSTEM_NAME,
description=f"{settings.SYSTEM_TITLE}",
version="6.0.0",
lifespan=lifespan,
)
# CORS
app.add_middleware(
CORSMiddleware,
allow_origins=["*"],
allow_credentials=True,
allow_methods=["*"],
allow_headers=["*"],
)
# TODO: Register routers (servers, auth, agent, scripts, credentials)
# from server.api.servers import router as servers_router
# from server.api.auth import router as auth_router
# from server.api.agent import router as agent_router
# from server.api.scripts import router as scripts_router
# from server.api.credentials import router as credentials_router
Nexus/tests/__init__.py
View File
Nexus/tests/conftest.py
+61
View File
@@ -0,0 +1,61 @@
"""Nexus — Test Configuration (pytest + async fixtures + Playwright)"""
import asyncio
import pytest
from httpx import AsyncClient, ASGITransport
from sqlalchemy.ext.asyncio import create_async_engine, async_sessionmaker, AsyncSession
from server.domain.models import Base
from server.main import app
from server.config import settings
# Use a test database (in-memory SQLite for unit tests, test MySQL for integration)
TEST_DATABASE_URL = "sqlite+aiosqlite:///test_nexus.db"
@pytest.fixture(scope="session")
def event_loop():
"""Create an instance of the default event loop for the test session"""
loop = asyncio.new_event_loop()
yield loop
loop.close()
@pytest.fixture(scope="session")
async def test_engine():
"""Create a test database engine"""
engine = create_async_engine(TEST_DATABASE_URL, echo=False)
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.create_all)
yield engine
async with engine.begin() as conn:
await conn.run_sync(Base.metadata.drop_all)
await engine.dispose()
@pytest.fixture(scope="function")
async def db_session(test_engine):
"""Create a fresh database session for each test"""
session_factory = async_sessionmaker(test_engine, class_=AsyncSession)
session = session_factory()
try:
yield session
finally:
await session.close()
@pytest.fixture(scope="function")
async def api_client(db_session):
"""Create an async HTTP client for API integration tests"""
# Override the get_async_session dependency
async def override_get_session():
yield db_session
app.dependency_overrides[...] = override_get_session # TODO: wire up
transport = ASGITransport(app=app)
client = AsyncClient(transport=transport, base_url="http://test")
yield client
await client.aclose()
app.dependency_overrides.clear()