refactor: slim config_profiles router
- Extract CRUD helpers to services/config/crud_service.py - Extract resolver/default logic to services/config/resolver_service.py - Slim config_profiles.py from 842 to 474 lines Quality gates: py_compile passes
This commit is contained in:
Developer
@@ -0,0 +1,208 @@
|
||||
"""Config profile resolver service functions."""
|
||||
|
||||
import logging
|
||||
import os
|
||||
import subprocess
|
||||
import uuid
|
||||
from typing import Any
|
||||
|
||||
from fastapi import HTTPException, status
|
||||
from sqlalchemy import select
|
||||
from sqlalchemy.ext.asyncio import AsyncSession
|
||||
|
||||
from src.models import ConfigProfile, SSHKey, UserConfig
|
||||
from src.services.shared.ssh_keys import _get_fernet
|
||||
from src.utils.git_url_parser import parse_git_url
|
||||
from src.schemas.config import ValidateGitUrlResponse
|
||||
|
||||
logger = logging.getLogger(__name__)
|
||||
|
||||
|
||||
async def resolve_default_profile(
|
||||
session: AsyncSession,
|
||||
user_id: uuid.UUID,
|
||||
project_id: uuid.UUID,
|
||||
tool_type_id: uuid.UUID,
|
||||
) -> dict:
|
||||
"""Resolve the default config profile for a project/tool combination."""
|
||||
query = (
|
||||
select(ConfigProfile)
|
||||
.where(ConfigProfile.user_id == user_id)
|
||||
.where(
|
||||
(ConfigProfile.project_id.is_(None) & ConfigProfile.tool_type_id.is_(None))
|
||||
| (ConfigProfile.project_id == project_id)
|
||||
| (ConfigProfile.tool_type_id == tool_type_id)
|
||||
| (
|
||||
(ConfigProfile.project_id == project_id)
|
||||
& (ConfigProfile.tool_type_id == tool_type_id)
|
||||
)
|
||||
)
|
||||
.order_by(ConfigProfile.created_at)
|
||||
)
|
||||
result = await session.execute(query)
|
||||
profiles = result.scalars().all()
|
||||
|
||||
if not profiles:
|
||||
return {"profile_id": None, "profile_name": None}
|
||||
|
||||
explicit_defaults = [p for p in profiles if p.is_default]
|
||||
|
||||
for p in explicit_defaults:
|
||||
if p.project_id == project_id and p.tool_type_id == tool_type_id:
|
||||
return {"profile_id": str(p.id), "profile_name": p.name}
|
||||
|
||||
for p in explicit_defaults:
|
||||
if p.project_id == project_id and p.tool_type_id is None:
|
||||
return {"profile_id": str(p.id), "profile_name": p.name}
|
||||
|
||||
for p in explicit_defaults:
|
||||
if p.project_id is None and p.tool_type_id == tool_type_id:
|
||||
return {"profile_id": str(p.id), "profile_name": p.name}
|
||||
|
||||
for p in explicit_defaults:
|
||||
if p.project_id is None and p.tool_type_id is None:
|
||||
return {"profile_id": str(p.id), "profile_name": p.name}
|
||||
|
||||
first = profiles[0]
|
||||
return {"profile_id": str(first.id), "profile_name": first.name}
|
||||
|
||||
|
||||
async def validate_git_url(
|
||||
session: AsyncSession,
|
||||
current_user_id: uuid.UUID,
|
||||
url: str,
|
||||
ssh_key_id: str | None,
|
||||
) -> ValidateGitUrlResponse:
|
||||
"""Validate a git remote URL and list available branches."""
|
||||
parse_result = parse_git_url(url)
|
||||
original_url = url.strip()
|
||||
url_to_check = parse_result.get("base_url") or original_url
|
||||
|
||||
if not url_to_check:
|
||||
return ValidateGitUrlResponse(
|
||||
valid=False,
|
||||
error=parse_result.get("message", "Invalid URL"),
|
||||
error_code=parse_result.get("error_code", "INVALID_URL"),
|
||||
)
|
||||
|
||||
if parse_result.get("needs_parsing") and url_to_check != original_url:
|
||||
return ValidateGitUrlResponse(
|
||||
valid=False,
|
||||
suggested_url=url_to_check,
|
||||
error=parse_result.get("message"),
|
||||
error_code=parse_result.get("error_code", "URL_NEEDS_PARSING"),
|
||||
)
|
||||
|
||||
env = None
|
||||
key_path = None
|
||||
if ssh_key_id:
|
||||
try:
|
||||
ssh_key_uuid = uuid.UUID(ssh_key_id)
|
||||
except ValueError:
|
||||
return ValidateGitUrlResponse(
|
||||
valid=False,
|
||||
error="Invalid SSH key ID format",
|
||||
error_code="INVALID_SSH_KEY",
|
||||
)
|
||||
|
||||
ssh_key = await session.get(SSHKey, ssh_key_uuid)
|
||||
if ssh_key is None or ssh_key.user_id != current_user_id:
|
||||
return ValidateGitUrlResponse(
|
||||
valid=False,
|
||||
error="SSH key not found or not authorized",
|
||||
error_code="SSH_KEY_NOT_FOUND",
|
||||
)
|
||||
|
||||
import tempfile
|
||||
|
||||
fernet = _get_fernet()
|
||||
private_key = fernet.decrypt(ssh_key.private_key_encrypted.encode()).decode()
|
||||
fd, key_path = tempfile.mkstemp(prefix="ssh_key_")
|
||||
try:
|
||||
os.write(fd, private_key.encode())
|
||||
finally:
|
||||
os.close(fd)
|
||||
os.chmod(key_path, 0o600)
|
||||
env = {
|
||||
"GIT_SSH_COMMAND": f"ssh -i {key_path} -o StrictHostKeyChecking=no -o UserKnownHostsFile=/dev/null"
|
||||
}
|
||||
|
||||
try:
|
||||
result = subprocess.run(
|
||||
["git", "ls-remote", "--heads", url_to_check],
|
||||
capture_output=True,
|
||||
text=True,
|
||||
timeout=30,
|
||||
env={**os.environ, **env} if env else None,
|
||||
)
|
||||
except subprocess.TimeoutExpired:
|
||||
if key_path and os.path.exists(key_path):
|
||||
os.unlink(key_path)
|
||||
return ValidateGitUrlResponse(
|
||||
valid=False,
|
||||
error="Remote repository check timed out",
|
||||
error_code="TIMEOUT",
|
||||
)
|
||||
except FileNotFoundError:
|
||||
if key_path and os.path.exists(key_path):
|
||||
os.unlink(key_path)
|
||||
return ValidateGitUrlResponse(
|
||||
valid=False,
|
||||
error="git command not found on server",
|
||||
error_code="GIT_NOT_FOUND",
|
||||
)
|
||||
finally:
|
||||
if key_path and os.path.exists(key_path):
|
||||
os.unlink(key_path)
|
||||
|
||||
if result.returncode != 0:
|
||||
stderr = result.stderr.strip()
|
||||
if (
|
||||
"could not resolve" in stderr.lower()
|
||||
or "unable to access" in stderr.lower()
|
||||
):
|
||||
error_msg = "Could not reach repository. Check the URL and network access."
|
||||
error_code = "UNREACHABLE"
|
||||
elif (
|
||||
"authentication" in stderr.lower() or "permission denied" in stderr.lower()
|
||||
):
|
||||
error_msg = (
|
||||
"Authentication failed. Provide an SSH key for private repositories."
|
||||
)
|
||||
error_code = "AUTH_FAILED"
|
||||
else:
|
||||
error_msg = f"Repository not accessible: {stderr[:200]}"
|
||||
error_code = "REMOTE_ERROR"
|
||||
return ValidateGitUrlResponse(
|
||||
valid=False,
|
||||
error=error_msg,
|
||||
error_code=error_code,
|
||||
)
|
||||
|
||||
branches: list[str] = []
|
||||
default_branch = "main"
|
||||
for line in result.stdout.strip().split("\n"):
|
||||
if not line.strip():
|
||||
continue
|
||||
parts = line.split()
|
||||
if len(parts) == 2:
|
||||
ref = parts[1]
|
||||
if ref.startswith("refs/heads/"):
|
||||
branch_name = ref[len("refs/heads/") :]
|
||||
branches.append(branch_name)
|
||||
if branch_name in ("main", "master"):
|
||||
default_branch = branch_name
|
||||
|
||||
if not branches:
|
||||
return ValidateGitUrlResponse(
|
||||
valid=False,
|
||||
error="No branches found in remote repository",
|
||||
error_code="NO_BRANCHES",
|
||||
)
|
||||
|
||||
return ValidateGitUrlResponse(
|
||||
valid=True,
|
||||
suggested_url=url_to_check if url_to_check != original_url else None,
|
||||
branches=branches,
|
||||
default_branch=default_branch,
|
||||
)
|
||||
Reference in New Issue
Block a user