As of January 1, 2020 this library no longer supports Python 2 on the latest released version. Library versions released prior to that date will continue to be available. For more information please visit Python 2 support on Google Cloud.

Helpers¶

General Helpers¶

Datetime Helpers¶

Helpers for datetime.

class google.api_core.datetime_helpers.DatetimeWithNanoseconds(*args, **kw)[source]¶

Bases: datetime.datetime

Track nanosecond in addition to normal datetime attrs.

Nanosecond can be passed only as a keyword argument.

classmethod from_rfc3339(stamp)[source]¶

Parse RFC3339-compliant timestamp, preserving nanoseconds.

Parameters: stamp (str) – RFC3339 stamp, with up to nanosecond precision
Returns: an instance matching the timestamp string
Return type: DatetimeWithNanoseconds
Raises: ValueError – if stamp does not match the expected format

classmethod from_timestamp_pb(stamp)[source]¶

Parse RFC3339-compliant timestamp, preserving nanoseconds.

Parameters: stamp (Timestamp) – timestamp message
Returns: an instance matching the timestamp message
Return type: DatetimeWithNanoseconds

property nanosecond¶

nanosecond precision.

Type: Read-only

rfc3339()[source]¶

Return an RFC3339-compliant timestamp.

Returns: Timestamp string according to RFC3339 spec.
Return type: (str)

timestamp_pb()[source]¶

Return a timestamp message.

Returns: Timestamp message
Return type: (Timestamp)

google.api_core.datetime_helpers.from_iso8601_date(value)[source]¶

Convert a ISO8601 date string to a date.

Parameters: value (str) – The ISO8601 date string.
Returns: A date equivalent to the date string.
Return type: datetime.date

google.api_core.datetime_helpers.from_iso8601_time(value)[source]¶

Convert a zoneless ISO8601 time string to a time.

Parameters: value (str) – The ISO8601 time string.
Returns: A time equivalent to the time string.
Return type: datetime.time

google.api_core.datetime_helpers.from_microseconds(value)[source]¶

Convert timestamp in microseconds since the unix epoch to datetime.

Parameters

value (float) – The timestamp to convert, in microseconds.

Returns

The datetime object equivalent to the timestamp in: UTC.

Return type

datetime.datetime

google.api_core.datetime_helpers.from_rfc3339(value)[source]¶

Convert an RFC3339-format timestamp to a native datetime.

Supported formats include those without fractional seconds, or with any fraction up to nanosecond precision.

Note

Python datetimes do not support nanosecond precision; this function therefore truncates such values to microseconds.

Parameters: value (str) – The RFC3339 string to convert.
Returns: The datetime object equivalent to the timestamp in UTC.
Return type: datetime.datetime
Raises: ValueError – If the timestamp does not match the RFC3339 regular expression.

google.api_core.datetime_helpers.from_rfc3339_nanos(value)¶

Convert an RFC3339-format timestamp to a native datetime.

Supported formats include those without fractional seconds, or with any fraction up to nanosecond precision.

Note

Python datetimes do not support nanosecond precision; this function therefore truncates such values to microseconds.

Parameters: value (str) – The RFC3339 string to convert.
Returns: The datetime object equivalent to the timestamp in UTC.
Return type: datetime.datetime
Raises: ValueError – If the timestamp does not match the RFC3339 regular expression.

google.api_core.datetime_helpers.to_microseconds(value)[source]¶

Convert a datetime to microseconds since the unix epoch.

Parameters: value (datetime.datetime) – The datetime to covert.
Returns: Microseconds since the unix epoch.
Return type: int

google.api_core.datetime_helpers.to_milliseconds(value)[source]¶

Convert a zone-aware datetime to milliseconds since the unix epoch.

Parameters: value (datetime.datetime) – The datetime to covert.
Returns: Milliseconds since the unix epoch.
Return type: int

google.api_core.datetime_helpers.to_rfc3339(value, ignore_zone=True)[source]¶

Convert a datetime to an RFC3339 timestamp string.

Parameters

value (datetime.datetime) – The datetime object to be converted to a string.
ignore_zone (bool) – If True, then the timezone (if any) of the datetime object is ignored and the datetime is treated as UTC.

Returns

The RFC3339 formatted string representing the datetime.

Return type

str

google.api_core.datetime_helpers.utcnow()[source]¶: A datetime.datetime.utcnow() alias to allow mocking in tests.

gRPC Helpers¶

Helpers for grpc.

class google.api_core.grpc_helpers.ChannelStub(responses=[])[source]¶

Bases: grpc.Channel

A testing stub for the grpc.Channel interface.

This can be used to test any client that eventually uses a gRPC channel to communicate. By passing in a channel stub, you can configure which responses are returned and track which requests are made.

For example:

channel_stub = grpc_helpers.ChannelStub()
client = FooClient(channel=channel_stub)

channel_stub.GetFoo.response = foo_pb2.Foo(name='bar')

foo = client.get_foo(labels=['baz'])

assert foo.name == 'bar'
assert channel_stub.GetFoo.requests[0].labels = ['baz']

Each method on the stub can be accessed and configured on the channel. Here’s some examples of various configurations:

# Return a basic response:

channel_stub.GetFoo.response = foo_pb2.Foo(name='bar')
assert client.get_foo().name == 'bar'

# Raise an exception:
channel_stub.GetFoo.response = NotFound('...')

with pytest.raises(NotFound):
    client.get_foo()

# Use a sequence of responses:
channel_stub.GetFoo.responses = iter([
    foo_pb2.Foo(name='bar'),
    foo_pb2.Foo(name='baz'),
])

assert client.get_foo().name == 'bar'
assert client.get_foo().name == 'baz'

# Use a callable

def on_get_foo(request):
    return foo_pb2.Foo(name='bar' + request.id)

channel_stub.GetFoo.response = on_get_foo

assert client.get_foo(id='123').name == 'bar123'

close()[source]¶: grpc.Channel.close implementation.

requests¶

A list of all requests made on this channel in order. The tuple is of method name, request message.

Type: Sequence[Tuple[str, protobuf.Message]]

stream_stream(method, request_serializer=None, response_deserializer=None, _registered_method=False)[source]¶: grpc.Channel.stream_stream implementation.

stream_unary(method, request_serializer=None, response_deserializer=None, _registered_method=False)[source]¶: grpc.Channel.stream_unary implementation.

subscribe(callback, try_to_connect=False)[source]¶: grpc.Channel.subscribe implementation.

unary_stream(method, request_serializer=None, response_deserializer=None, _registered_method=False)[source]¶: grpc.Channel.unary_stream implementation.

unary_unary(method, request_serializer=None, response_deserializer=None, _registered_method=False)[source]¶: grpc.Channel.unary_unary implementation.

unsubscribe(callback)[source]¶: grpc.Channel.unsubscribe implementation.

google.api_core.grpc_helpers.apply_channel_interceptors(channel: grpc.Channel, interceptors: Optional[Sequence[Union[grpc.UnaryUnaryClientInterceptor, grpc.UnaryStreamClientInterceptor, grpc.StreamUnaryClientInterceptor, grpc.StreamStreamClientInterceptor, Callable[[grpc.Channel], grpc.Channel]]]] = None) → grpc.Channel[source]¶

Applies client interceptors or channel-intercepting callables to a gRPC channel.

Executes in reverse order so the first interceptor in the sequence becomes the outermost layer on outbound requests and the innermost layer on inbound responses, aligning with the behavior of grpc.intercept_channel.

Parameters

channel (grpc.Channel) – The channel to intercept.
interceptors (Optional[Sequence[Union[ClientInterceptor, Callable[[grpc.Channel], grpc.Channel]]]]) – Additional interceptors (or callables that apply interceptors) to apply to the gRPC channel.

Returns

The intercepted channel, or the original channel if no: interceptors were provided.

Return type

grpc.Channel

Raises

TypeError – If an item in interceptors is neither a gRPC ClientInterceptor nor a Callable[[Channel], Channel].

google.api_core.grpc_helpers.create_channel(target, credentials=None, scopes=None, ssl_credentials=None, credentials_file=None, quota_project_id=None, default_scopes=None, default_host=None, compression=None, attempt_direct_path: Optional[bool] =