(env) sirneij@pop-os ~/D/P/T/django_excel (main)> pip install -r requirements.txt

(env) sirneij@pop-os ~/D/P/T/django_excel (main)> python manage.py migrate

(env) sirneij@pop-os ~/D/P/T/django_excel (main)> python manage.py run

(env) sirneij@pop-os ~/D/P/T/django_excel (main)> py.test --nomigrations --reuse-db -W error::RuntimeWarning --cov=core --cov-report=html tests/

# django_excel/core/tasks.py
...

@shared_task
def populate_googlesheet_with_coins_data() -> None:
    """Populate Googlesheet with the coin data from the database."""
    response = requests.get(settings.GOOGLE_API_SERVICE_KEY_URL)
    with open('core/djangoexcel.json', 'wb') as file:
        file.write(response.content)
    service_account_file = 'core/djangoexcel.json'
    creds = service_account.Credentials.from_service_account_file(
        service_account_file, scopes=settings.GOOGLE_API_SCOPE
    )
    service = build('sheets', 'v4', credentials=creds)
    sheet = service.spreadsheets()

    sheet_metadata_values = sheet.get(spreadsheetId=settings.SPREADSHEET_ID).execute()
    csheets = sheet_metadata_values.get('sheets', '')
    datetime_format = '%a %b %d %Y %Hh%Mm'
    if csheets:
        for csheet in csheets:
            sheet_title = csheet.get('properties', {}).get('title', '')
            date_segment_of_the_title = ' '.join(sheet_title.split(' ')[0:5]).strip()
            parsed_datetime: Optional[Any] = None
            try:
                parsed_datetime = datetime.strptime(date_segment_of_the_title, datetime_format)
            except ValueError as err:
                print(err)
            now = timezone.now().strftime(datetime_format)
            if (
                parsed_datetime
                and (datetime.strptime(now, datetime_format) - parsed_datetime).seconds
                > settings.SPREADSHEET_TAB_EXPIRY
            ):
                sheet_id = csheet.get('properties', {}).get('sheetId', 0)
                batch_update_request_body = {'requests': [{'deleteSheet': {'sheetId': sheet_id}}]}
                sheet.batchUpdate(spreadsheetId=settings.SPREADSHEET_ID, body=batch_update_request_body).execute()

    coin_queryset = Coins.objects.all().order_by('rank')
    coin_data_list: list[Any] = [
        [
            'Name',
            'Symbol',
            'Rank',
            'Current price',
            'Price change',
            'Market cap',
            'Total supply',
        ]
    ]
    for coin in coin_queryset:
        coin_data_list.append(
            [
                coin.name,
                f'{coin.symbol}'.upper(),
                coin.rank,
                str(currency(coin.current_price)),
                str(currency(coin.price_change_within_24_hours)),
                str(currency(coin.market_cap)),
                str(coin.total_supply),
            ]
        )

    new_sheet_title = f'{timezone.now().strftime(datetime_format)} Coin data'
    batch_update_request_body = {
        'requests': [
            {
                'addSheet': {
                    'properties': {
                        'title': new_sheet_title,
                        'tabColor': {'red': 0.968627451, 'green': 0.576470588, 'blue': 0.101960784},
                        'gridProperties': {'rowCount': len(coin_data_list), 'columnCount': 7},
                    }
                }
            }
        ]
    }
    sheet.batchUpdate(spreadsheetId=settings.SPREADSHEET_ID, body=batch_update_request_body).execute()
    sheet.values().append(
        spreadsheetId=settings.SPREADSHEET_ID,
        range=f"'{new_sheet_title}'!A1:G1",
        valueInputOption='USER_ENTERED',
        body={'values': coin_data_list},
    ).execute()

함수의 처음 9줄은 익숙할 것입니다. 인증을 위한 자격 증명을 만들려고 시도한 다음 스프레드시트가 인스턴스화되었습니다. 다음 줄로 이동하여 스프레드시트에서 모든 메타 데이터를 가져온 다음 스프레드시트에서 현재 사용 가능한 모든 시트를 가져옵니다. 그런 다음 사용할 날짜/시간 형식을 정의합니다. 이 형식은 datetime을 예를 들어 Fri Jun 17 07h34m 와 같이 출력하므로 시트의 제목은 Fri Jun 17 07h34m Coin data 와 같습니다. 그런 다음 이 시트를 반복하고 시트의 제목에서 날짜/시간 세그먼트를 구문 분석합니다. 이것의 목적은 오래된 모든 시트(이 경우 6 mimutes (360 seconds) 보다 오래된 모든 시트)를 삭제할 수 있도록 해당 시트가 생성된 날짜 시간을 아는 것입니다. 시트가 6분보다 오래된 경우 문서화된 대로here 자동으로 삭제할 수 있도록 해당 iD를 Google 스프레드시트deleteSheet 요청에 전달했습니다. 이 블록:

...
new_sheet_title = f'{timezone.now().strftime(datetime_format)} Coin data'
batch_update_request_body = {
    'requests': [
        {
            'addSheet': {
                'properties': {
                    'title': new_sheet_title,
                    'tabColor': {'red': 0.968627451, 'green': 0.576470588, 'blue': 0.101960784},
                    'gridProperties': {'rowCount': len(coin_data_list), 'columnCount': 7},
                }
            }
        }
    ]
}

요청이 이루어진 날짜와 시간을 기반으로 고유한 제목을 생성하고 이 제목을 문서화된 대로 addSheet 요청에 전달합니다 here . 다른 탭을 만들려면 고유한 제목이 필요합니다. 사실 제목이 중복되면 오류가 발생합니다. 다음 줄은 시트를 만들고 다음 줄에 데이터가 추가되었습니다. 그게 다야!!!

2단계: tests/core/test_tasks.py 업데이트

논리가 작동하는지 확인하고 이전에 작성된 테스트를 업데이트하겠습니다.

# tests/core/test_tasks.py
...
    def test_populate_googlesheet_with_coins_data(self):
        """Test populate_googlesheet_with_coins_data."""

        Coins.objects.create(
            name='bitcoin', symbol='btc', current_price=12000000, price_change_within_24_hours=500, market_cap=210000000
        )
        Coins.objects.create(
            name='etherum', symbol='eth', current_price=12000000, price_change_within_24_hours=500, market_cap=210000000
        )
        Coins.objects.create(
            name='xrp', symbol='xrp', current_price=12000000, price_change_within_24_hours=500, market_cap=210000000
        )

        with patch('core.tasks.build') as mock_build:
            with patch('core.tasks.service_account.Credentials') as mock_service_acount_credentials:
                mock_service_acount_credentials.from_service_account_info.return_value = '123'
                datetime_format = '%a %b %d %Y %Hh%Mm'
                mock_build.return_value.spreadsheets.return_value.get.return_value.execute.return_value = {
                    'spreadsheetId': '1AFNyUKcqgwO-CCXRubcIALOC74yfV716Q5q57Ojjicc',
                    'sheets': [
                        {
                            'properties': {
                                'sheetId': 0,
                                'title': 'Coins',
                            }
                        },
                        {
                            'properties': {
                                'sheetId': 1305535527,
                                'title': f'{timezone.now().strftime(datetime_format)} Coin data',
                            }
                        },
                    ],
                    'spreadsheetUrl': 'https://docs.google.com/spreadsheets/d/1AFNyUKcqgwO-CCXRubcIALOC74yfV716Q5q57Ojjicc/edit',
                }

                today_datetime_now = timezone.now() + timedelta(minutes=7)
                with patch('django.utils.timezone.now', return_value=today_datetime_now):
                    populate_googlesheet_with_coins_data()

        mock_build.assert_called_once()

스프레드시트 API에 대한 get 요청에서 받은 응답을 조롱해야 했기 때문에 이 블록은 다음과 같습니다.

mock_build.return_value.spreadsheets.return_value.get.return_value.execute.return_value = {
        'spreadsheetId': '1AFNyUKcqgwO-CCXRubcIALOC74yfV716Q5q57Ojjicc',
        'sheets': [
            {
                'properties': {
                    'sheetId': 0,
                    'title': 'Coins',
                }
            },
            {
                'properties': {
                    'sheetId': 1305535527,
                    'title': f'{timezone.now().strftime(datetime_format)} Coin data',
                }
            },
        ],
        'spreadsheetUrl': 'https://docs.google.com/spreadsheets/d/1AFNyUKcqgwO-CCXRubcIALOC74yfV716Q5q57Ojjicc/edit',
    }

6분 이상의 시간을 캡처하기 위해 미래의 특정 시간을 제공하기 위해 timezone.now()를 조롱했습니다. 테스트를 실행할 때 모든 함수의 코드를 다루어야 합니다...

아웃트로

이 기사를 즐겼습니까? contacting me for a job, something worthwhile or buying a coffee ☕을 고려하십시오. 에 연결/팔로우할 수도 있습니다. 또한 더 넓은 범위를 위해 공유하는 데 도움이된다면 나쁘지 않습니다. 나는 감사 할 것입니다...