💡
djangoでgithub連携の実装
url
path("api/v1/github/", include("app.github_sync.urls")),
path("api/v1/auth/social/", include("social_django.urls", namespace="social")),
views
from rest_framework import viewsets, status
from rest_framework.decorators import action
from rest_framework.response import Response
from rest_framework.permissions import IsAuthenticated
from django.shortcuts import get_object_or_404
from .models import GitHubRepository, GitHubSync, GitHubWebhook
from .serializers import (
GitHubRepositorySerializer,
GitHubSyncSerializer,
GitHubSyncCreateSerializer,
GitHubSyncStatusSerializer,
GitHubWebhookSerializer
)
from .services import GitHubSyncService
from app.posts.models import Post
from social_django.models import UserSocialAuth
import logging
logger = logging.getLogger(name)
class GitHubRepositoryViewSet(viewsets.ModelViewSet):
"""GitHub リポジトリ管理ビュー"""
serializer_class = GitHubRepositorySerializer
permission_classes = [IsAuthenticated]
def get_queryset(self):
return GitHubRepository.objects.filter(user=self.request.user)
def perform_create(self, serializer):
serializer.save(user=self.request.user)
@action(detail=False, methods=['get'])
def available_repos(self, request):
"""利用可能なGitHubリポジトリ一覧を取得"""
try:
# ユーザーのGitHubアクセストークンを取得
social_auth = UserSocialAuth.objects.get(
user=request.user,
provider='github'
)
access_token = social_auth.extra_data.get('access_token')
if not access_token:
return Response(
{'error': 'GitHub access token not found'},
status=status.HTTP_400_BAD_REQUEST
)
# GitHubサービスを初期化
service = GitHubSyncService(access_token)
# リポジトリ一覧を取得
repos = service.get_user_repos()
# 既に連携済みのリポジトリを除外
connected_repos = GitHubRepository.objects.filter(
user=request.user
).values_list('repo_full_name', flat=True)
available_repos = [
repo for repo in repos
if repo['full_name'] not in connected_repos
]
return Response(available_repos)
except UserSocialAuth.DoesNotExist:
return Response(
{'error': 'GitHub account not connected'},
status=status.HTTP_400_BAD_REQUEST
)
except Exception as e:
logger.error(f"Failed to get available repos: {e}")
return Response(
{'error': str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
@action(detail=True, methods=['post'])
def sync_all(self, request, pk=None):
"""リポジトリの全記事を同期"""
repository = self.get_object()
try:
# GitHubアクセストークンを取得
social_auth = UserSocialAuth.objects.get(
user=request.user,
provider='github'
)
access_token = social_auth.extra_data.get('access_token')
if not access_token:
return Response(
{'error': 'GitHub access token not found'},
status=status.HTTP_400_BAD_REQUEST
)
# 同期処理
service = GitHubSyncService(access_token)
synced_count = 0
failed_count = 0
for sync in repository.synced_posts.all():
if service.sync_post_to_github(sync):
synced_count += 1
else:
failed_count += 1
return Response({
'synced': synced_count,
'failed': failed_count
})
except Exception as e:
logger.error(f"Failed to sync repository: {e}")
return Response(
{'error': str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
class GitHubSyncViewSet(viewsets.ModelViewSet):
"""GitHub同期管理ビュー"""
serializer_class = GitHubSyncSerializer
permission_classes = [IsAuthenticated]
def get_queryset(self):
return GitHubSync.objects.filter(
post__author=self.request.user
).select_related('post', 'repository')
def create(self, request):
"""新規同期設定を作成"""
serializer = GitHubSyncCreateSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
try:
# 記事とリポジトリを取得
post = get_object_or_404(
Post,
id=serializer.validated_data['post_id'],
author=request.user
)
repository = get_object_or_404(
GitHubRepository,
id=serializer.validated_data['repository_id'],
user=request.user
)
# ファイルパスを生成
file_path = serializer.validated_data.get('file_path')
if not file_path:
# GitHubアクセストークンを取得
social_auth = UserSocialAuth.objects.get(
user=request.user,
provider='github'
)
access_token = social_auth.extra_data.get('access_token')
service = GitHubSyncService(access_token)
file_path = service.generate_file_path(post, repository)
# 同期設定を作成
sync = GitHubSync.objects.create(
post=post,
repository=repository,
file_path=file_path
)
# 初回同期を実行
service = GitHubSyncService(access_token)
service.sync_post_to_github(sync)
# レスポンス用にシリアライズ
response_serializer = GitHubSyncSerializer(sync)
return Response(
response_serializer.data,
status=status.HTTP_201_CREATED
)
except Exception as e:
logger.error(f"Failed to create sync: {e}")
return Response(
{'error': str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
@action(detail=True, methods=['post'])
def sync_status(self, request, pk=None):
"""同期状態を更新"""
sync = self.get_object()
serializer = GitHubSyncStatusSerializer(data=request.data)
serializer.is_valid(raise_exception=True)
try:
# GitHubアクセストークンを取得
social_auth = UserSocialAuth.objects.get(
user=request.user,
provider='github'
)
access_token = social_auth.extra_data.get('access_token')
if not access_token:
return Response(
{'error': 'GitHub access token not found'},
status=status.HTTP_400_BAD_REQUEST
)
service = GitHubSyncService(access_token)
action = serializer.validated_data['action']
if action == 'pull':
# GitHubから取得
success = service.sync_post_from_github(sync)
elif action == 'push':
# GitHubへプッシュ
success = service.sync_post_to_github(sync)
elif action == 'resolve':
# コンフリクト解決（強制プッシュ）
if serializer.validated_data.get('force'):
success = service.sync_post_to_github(sync)
else:
return Response(
{'error': 'Force flag required for conflict resolution'},
status=status.HTTP_400_BAD_REQUEST
)
else:
return Response(
{'error': 'Invalid action'},
status=status.HTTP_400_BAD_REQUEST
)
if success:
sync.refresh_from_db()
response_serializer = GitHubSyncSerializer(sync)
return Response(response_serializer.data)
else:
return Response(
{'error': sync.last_error},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
except Exception as e:
logger.error(f"Failed to update sync status: {e}")
return Response(
{'error': str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
@action(detail=True, methods=['get'])
def check_status(self, request, pk=None):
"""同期状態をチェック"""
sync = self.get_object()
try:
# GitHubアクセストークンを取得
social_auth = UserSocialAuth.objects.get(
user=request.user,
provider='github'
)
access_token = social_auth.extra_data.get('access_token')
if not access_token:
return Response(
{'error': 'GitHub access token not found'},
status=status.HTTP_400_BAD_REQUEST
)
service = GitHubSyncService(access_token)
status_result = service.check_sync_status(sync)
# ステータスを更新
sync.sync_status = status_result
sync.save()
return Response({
'status': status_result,
'last_synced_at': sync.last_synced_at,
'last_error': sync.last_error
})
except Exception as e:
logger.error(f"Failed to check sync status: {e}")
return Response(
{'error': str(e)},
status=status.HTTP_500_INTERNAL_SERVER_ERROR
)
Discussion