Typing just does not work with that function

[jefer94]

Does this issue occur when all extensions are disabled?: I don't know

Version: 1.94.0-insider (user setup)
Commit: c4efe1dc9eec4914f3076b2d954fe4fe174a5820
Date: 2024-09-17T05:03:51.290Z
Electron: 30.4.0
ElectronBuildId: 10073054
Chromium: 124.0.6367.243
Node.js: 20.15.1
V8: 12.4.254.20-electron.0
OS: Windows_NT x64 10.0.26120

Windows 11 Home 24H2 26120.1542
Windows Feature Experience Pack 1000.26100.18.0

Linux Jeferson 5.15.146.1-microsoft-standard-WSL2 #1 SMP Thu Jan 11 04:09:03 UTC 2024 x86_64 GNU/Linux

Steps to Reproduce:

import os.path
from typing import Optional, TypedDict, Unpack, AsyncIterator
from google.apps.meet_v2 import ConferenceRecord

import google.apps.meet_v2.services.conference_records_service.pagers as pagers
from asgiref.sync import async_to_sync
from google.apps import meet_v2
from google.apps.meet_v2.types import Space
from google.protobuf.field_mask_pb2 import FieldMask
from google.oauth2.credentials import Credentials


import pickle
from google.auth.transport.requests import Request

__all__ = ["GoogleMeet"]


class CreateSpaceRequest(TypedDict):
    space: Space


class EndActiveConferenceRequest(TypedDict):
    name: str


class GetConferenceRecordRequest(TypedDict):
    name: str


class GetParticipantRequest(TypedDict):
    name: str


class GetParticipantSessionRequest(TypedDict):
    name: str


class GetRecordingRequest(TypedDict):
    name: str


class GetSpaceRequest(TypedDict):
    name: str


class UpdateSpaceRequest(TypedDict):
    space: Space
    update_mask: FieldMask


class GetTranscriptRequest(TypedDict):
    name: str


class ListConferenceRecordsRequest(TypedDict):
    page_size: int
    page_token: str
    filter: str  # in EBNF format, space.meeting_code, space.name, start_time and end_time


class ListRecordingsRequest(TypedDict):
    parent: str
    page_size: int
    page_token: str


class ListParticipantSessionsRequest(TypedDict):
    parent: str
    page_size: int
    page_token: str
    filter: str  # in EBNF format, start_time and end_time


class ListTranscriptsRequest(TypedDict):
    parent: str
    page_size: int
    page_token: str


class ListParticipantsRequest(TypedDict):
    parent: str
    page_size: int
    page_token: str
    filter: str  # in EBNF format, start_time and end_time


# Scopes for Google Calendar API (used for creating Google Meet links)
# https://www.googleapis.com/auth/meetings.space.created
# https://www.googleapis.com/auth/meetings.space.readonly
# SCOPES = [
#     "google.apps.meet.v2.SpacesService.CreateSpace",
#     "google.apps.meet.v2.SpacesService.GetSpace",
# ]
TOKEN_FILE_NAME = "google_cloud_oauth_token.pickle"
GOOGLE_CLIENT_SECRET = "client_secret.json"


class GoogleMeet:
    _spaces_service_client: Optional[meet_v2.SpacesServiceAsyncClient]
    _conference_records_service_client: Optional[meet_v2.ConferenceRecordsServiceAsyncClient]

    def __init__(self, token: str, refresh_token: Optional[str] = None):
        self._credentials = self._get_credentials(token, refresh_token)
        self._spaces_service_client = None
        self._conference_records_service_client = None

    def _get_credentials(
        self, token: Optional[str] = None, refresh_token: Optional[str] = None
    ) -> Optional[Credentials]:
        creds = None

        if token:
            creds = Credentials(
                token=token,
                refresh_token=refresh_token,
                token_uri="https://oauth2.googleapis.com/token",
                client_id=os.getenv("GOOGLE_CLIENT_ID"),
                client_secret=os.getenv("GOOGLE_SECRET"),
            )
            creds.refresh(Request())
        elif os.path.exists(TOKEN_FILE_NAME):
            with open(TOKEN_FILE_NAME, "rb") as token:
                creds = pickle.load(token)
                creds.refresh(Request())

        # If there are no valid credentials available, raise an exception
        if not creds or not creds.valid:
            if creds and creds.expired and creds.refresh_token:
                creds.refresh(Request())
            else:
                raise Exception("Invalid or expired credentials. Please provide a valid token.")

        return creds

    async def spaces_service_client(self):
        if self._spaces_service_client is None:
            self._spaces_service_client = meet_v2.SpacesServiceAsyncClient(credentials=self._credentials)

        return self._spaces_service_client

    async def conference_records_service_client(self):
        if self._conference_records_service_client is None:
            self._conference_records_service_client = meet_v2.ConferenceRecordsServiceAsyncClient(
                credentials=self._credentials
            )

        return self._conference_records_service_client

    async def acreate_space(self, **kwargs: Unpack[CreateSpaceRequest]) -> meet_v2.Space:

        # Create a client
        client = await self.spaces_service_client()

        # Initialize request argument(s)
        request = meet_v2.CreateSpaceRequest(**kwargs)

        # Make the request
        return await client.create_space(request=request)

    @async_to_sync
    async def create_space(self, **kwargs: Unpack[CreateSpaceRequest]) -> meet_v2.Space:
        return await self.acreate_space(**kwargs)

    async def aget_space(self, **kwargs: Unpack[GetSpaceRequest]) -> meet_v2.Space:

        # Create a client
        client = await self.spaces_service_client()

        # Initialize request argument(s)
        request = meet_v2.GetSpaceRequest(**kwargs)

        # Make the request
        response = await client.get_space(request=request)

        # Handle the response
        print(response)

    @async_to_sync
    async def get_space(self, **kwargs: Unpack[GetSpaceRequest]) -> meet_v2.Space:
        return await self.aget_space(**kwargs)

    async def aupdate_space(self, **kwargs: Unpack[UpdateSpaceRequest]) -> meet_v2.Space:
        # Create a client
        client = await self.spaces_service_client()

        # Initialize request argument(s)
        request = meet_v2.UpdateSpaceRequest(**kwargs)

        # Make the request
        response = await client.update_space(request=request)

        # Handle the response
        print(response)

    @async_to_sync
    async def update_space(self, **kwargs: Unpack[UpdateSpaceRequest]) -> meet_v2.Space:
        return await self.aupdate_space(**kwargs)

    async def aend_active_conference(self, **kwargs: Unpack[EndActiveConferenceRequest]) -> None:
        # Create a client
        client = await self.spaces_service_client()

        # Initialize request argument(s)
        request = meet_v2.EndActiveConferenceRequest(**kwargs)

        # Make the request
        return await client.end_active_conference(request=request)

    @async_to_sync
    async def end_active_conference(self, **kwargs: Unpack[EndActiveConferenceRequest]) -> None:
        return await self.aend_active_conference(**kwargs)

    async def alist_participants(self, **kwargs: Unpack[ListParticipantsRequest]) -> pagers.ListParticipantsAsyncPager:
        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.ListParticipantsRequest(**kwargs)

        # Make the request
        return await client.list_participants(request=request)

    async def aget_participant(self, **kwargs: Unpack[GetParticipantRequest]) -> meet_v2.Participant:
        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.GetParticipantRequest(**kwargs)

        # Make the request
        return await client.get_participant(request=request)

    @async_to_sync
    async def get_participant(self, **kwargs: Unpack[GetParticipantRequest]) -> meet_v2.Participant:
        return await self.aget_participant(**kwargs)

    async def alist_participant_sessions(
        self, **kwargs: Unpack[ListParticipantSessionsRequest]
    ) -> pagers.ListParticipantSessionsAsyncPager:
        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.ListParticipantSessionsRequest(**kwargs)

        # Make the request
        return await client.list_participant_sessions(request=request)

    async def aget_participant_session(
        self, **kwargs: Unpack[GetParticipantSessionRequest]
    ) -> meet_v2.ParticipantSession:
        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.GetParticipantSessionRequest(**kwargs)

        # Make the request
        return await client.get_participant_session(request=request)

    @async_to_sync
    async def get_participant_session(
        self, **kwargs: Unpack[GetParticipantSessionRequest]
    ) -> meet_v2.ParticipantSession:
        return await self.aget_participant_session(**kwargs)

    async def alist_recordings(self, **kwargs: Unpack[ListRecordingsRequest]) -> pagers.ListRecordingsAsyncPager:
        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.ListRecordingsRequest(**kwargs)

        # Make the request
        return await client.list_recordings(request=request)

    async def aget_recording(self, **kwargs: Unpack[GetRecordingRequest]) -> meet_v2.Recording:
        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.GetRecordingRequest(**kwargs)

        # Make the request
        response = await client.get_recording(request=request)

        # Handle the response
        print(response)

    @async_to_sync
    async def get_recording(self, **kwargs: Unpack[GetRecordingRequest]) -> meet_v2.Recording:
        return await self.aget_recording(**kwargs)

    async def alist_transcripts(self, **kwargs: Unpack[ListTranscriptsRequest]) -> pagers.ListTranscriptsAsyncPager:
        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.ListTranscriptsRequest(**kwargs)

        # Make the request
        return await client.list_transcripts(request=request)

    async def aget_transcript(self, **kwargs: Unpack[GetTranscriptRequest]) -> meet_v2.Transcript:
        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.GetTranscriptRequest(**kwargs)

        # Make the request
        response = await client.get_transcript(request=request)

        # Handle the response
        print(response)

    @async_to_sync
    async def get_transcript(self, **kwargs: Unpack[GetTranscriptRequest]) -> meet_v2.Transcript:
        return await self.aget_transcript(**kwargs)

    async def alist_conference_records(
        self,
        **kwargs: Unpack[ListConferenceRecordsRequest]
        # ) -> AsyncIterator[ConferenceRecord]:
    ) -> pagers.ListConferenceRecordsAsyncPager:

        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.ListConferenceRecordsRequest(**kwargs)

        # Make the request
        pager = await client.list_conference_records(request=request)
        return pager

        # Yield each conference record
        async for record in pager:
            yield record

    async def aget_conference_record(self, **kwargs: Unpack[GetConferenceRecordRequest]) -> meet_v2.ConferenceRecord:
        # Create a client
        client = await self.conference_records_service_client()

        # Initialize request argument(s)
        request = meet_v2.GetConferenceRecordRequest(**kwargs)

        # Make the request
        return await client.get_conference_record(request=request)

    @async_to_sync
    async def get_conference_record(self, **kwargs: Unpack[GetConferenceRecordRequest]) -> meet_v2.ConferenceRecord:
        return await self.aget_conference_record(**kwargs)

[mjbvz]

/extPython

[vs-code-engineering]

It looks like this is caused by the Python extension. Please file the issue to the Python extension repository. Make sure to check their issue reporting template and provide them relevant information such as the extension version you're using. See also our issue reporting guidelines for more information.

Happy Coding!