fix: indentation gitea_client + request defaults
FlowDeck CI / test (push) Failing after 4s
FlowDeck CI / docker (push) Has been skipped

This commit is contained in:
2026-07-13 14:51:58 -04:00
parent 9f667ae9e0
commit 5baae598bf
2 changed files with 14 additions and 21 deletions
+3 -3
View File
@@ -43,7 +43,7 @@ async def list_projects(request: Request, org: str = ""):
# ── Repo tree ──
@router.get("/projects/{owner}/{repo}/tree")
async def repo_tree(owner: str, repo: str, path: str = "", request: Request = None):
async def repo_tree(owner: str, repo: str, path: str = "", request: Request):
"""Get contents of a repo directory (one level)."""
gitea = _require_gitea(request)
try:
@@ -64,7 +64,7 @@ async def repo_tree(owner: str, repo: str, path: str = "", request: Request = No
# ── File content ──
@router.get("/projects/{owner}/{repo}/file")
async def get_file(owner: str, repo: str, path: str, request: Request = None):
async def get_file(owner: str, repo: str, path: str, request: Request):
"""Get file content (decoded)."""
gitea = _require_gitea(request)
try:
@@ -117,7 +117,7 @@ async def delete_file(owner: str, repo: str, request: Request):
# ── Labels ──
@router.get("/projects/{owner}/{repo}/labels")
async def get_labels(owner: str, repo: str, request: Request = None):
async def get_labels(owner: str, repo: str, request: Request):
"""Get labels for a repo (mapped to tags)."""
gitea = _require_gitea(request)
try:
+11 -18
View File
@@ -300,13 +300,9 @@ class GiteaClient:
return data
# ── singleton ──
gitea = GiteaClient()
# ── repo contents ──
async def get_repo_contents(self, owner: str, repo: str, path: str = "") -> list[dict]:
async def get_repo_contents(self, owner, repo, path=""):
"""Get contents of a repo directory (lazy: one level)."""
url = f"{self._base}/repos/{owner}/{repo}/contents"
if path:
@@ -315,25 +311,22 @@ gitea = GiteaClient()
cached = self._cached(cache_key)
if cached:
return cached
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.get(url, headers=self._headers)
resp.raise_for_status()
data = resp.json()
# Ensure data is a list (single file returns dict)
if isinstance(data, dict):
data = [data]
self._set_cache(cache_key, data)
return data
async def get_file_content(self, owner: str, repo: str, path: str) -> str:
async def get_file_content(self, owner, repo, path):
"""Get raw file content (decoded from base64)."""
import base64
cache_key = f"file:{owner}:{repo}:{path}"
cached = self._cached(cache_key)
if cached:
return cached
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.get(
f"{self._base}/repos/{owner}/{repo}/contents/{path}",
@@ -350,10 +343,8 @@ gitea = GiteaClient()
self._set_cache(cache_key, content)
return content
async def create_or_update_file(
self, owner: str, repo: str, path: str, content: str, message: str, sha: str | None = None
) -> dict:
"""Create or update a file in the repo. Requires `sha` for updates."""
async def create_or_update_file(self, owner, repo, path, content, message, sha=None):
"""Create or update a file. Requires `sha` for updates."""
import base64
body = {
"content": base64.b64encode(content.encode("utf-8")).decode("utf-8"),
@@ -369,14 +360,13 @@ gitea = GiteaClient()
)
resp.raise_for_status()
data = resp.json()
# Invalidate caches
parent = "/".join(path.split("/")[:-1])
for p in (parent, path.rsplit("/", 1)[0] if "/" in path else ""):
self._cache.pop(f"contents:{owner}:{repo}:{p}", None)
self._cache.pop(f"file:{owner}:{repo}:{path}", None)
return data
async def delete_file(self, owner: str, repo: str, path: str, sha: str, message: str) -> dict:
async def delete_file(self, owner, repo, path, sha, message):
"""Delete a file from the repo."""
async with httpx.AsyncClient(timeout=15) as client:
resp = await client.delete(
@@ -391,16 +381,19 @@ gitea = GiteaClient()
self._cache.pop(f"file:{owner}:{repo}:{path}", None)
return {}
def _invalidate_tree_cache(self, owner: str, repo: str) -> None:
"""Clear all cached contents for a repo."""
def _invalidate_tree_cache(self, owner, repo):
keys = [k for k in self._cache if k.startswith(f"contents:{owner}:{repo}:") or k.startswith(f"file:{owner}:{repo}:")]
for k in keys:
del self._cache[k]
# ── singleton ──
gitea = GiteaClient()
# ── Helper: get per-user GiteaClient ──
def get_user_gitea_client(request) -> GiteaClient | None:
def get_user_gitea_client(request):
"""Return a GiteaClient authenticated with the user's OAuth token, or None."""
from app.auth.session import SessionManager
user = SessionManager.decode_session(request.cookies.get("flowdeck_session", ""))