`tlc.service.object_service`#

The 3LC Object Service.

Module Contents#

Classes#

Class	Description
`LitestarStateConstants`	Constants for the Litestar state.
`TLCObject`	In-flight representation of a TLCObject.
`TLCPatchOptions`	TLC patch request.
`TLCPatchRequest`	In-flight representation of a patch request for a TLCObject.
`RollbackDeleteContext`	A context manager for rollback object creation without interfering with InsufficientCredits.
`ObjectServiceFeatures`
`ObjectServiceUserInfo`
`DashboardAnnotations`
`ETagMiddleware`	A middleware that may serve 304 not modified responses based on ETag headers.
`DashboardKeyGuard`
`ObjectRoutesController`	Controller for all object-related routes
`ExternalDataRoutesController`	Controller for all external data-related routes
`TLCCustomLoggingMiddleware`	Custom middleware to log object service requests and responses.
`NGrokOutputAdaptor`	Helper class to format output from NGrokHelper for the Object Service.

Functions#

Function	Description
`internal_server_error_handler`	Catch-all for application errors.
`insufficient_credits_handler`	Handler for insufficient credits.
`resolve_cache_timeout`
`get_ip_addresses`
`get_running_urls`
`format_yaml_for_logging`
`open_in_web_browser`
`open_dashboard_in_web_browser`
`startup`	Setup HTTP client for connecting to 3LC Data Service
`shutdown`	Perform any required cleanup before terminating the application
`root`	Root endpoint of the service
`get_status`	Returns status of the service
`get_last_lru_stats`
`status`	Returns status of the service
`create_litestar_app`

Data#

Data	Description
`last_lru_stats`
`T`

API#

class tlc.service.object_service.LitestarStateConstants#

Constants for the Litestar state.

HOST_IP = HostIP#

OBJECT_SERVICE_RUNNING_URLS = ObjectServiceRunningsURLs#

NGROK_OBJECT_SERVICE_URL = NGrokObjectServiceURL#

tlc.service.object_service.internal_server_error_handler(request: litestar.Request, exception: Exception) → litestar.Response#: Catch-all for application errors.

tlc.service.object_service.insufficient_credits_handler(request: litestar.Request, exception: tlcsaas.transaction.InsufficientCredits) → litestar.Response#: Handler for insufficient credits.

tlc.service.object_service.resolve_cache_timeout() → int#

class tlc.service.object_service.TLCObject(/, **data: typing.Any)#

Bases: pydantic.BaseModel

In-flight representation of a TLCObject.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

type: str = None#

url: Optional[str] = None#

model_config = ConfigDict(...)#

class tlc.service.object_service.TLCPatchOptions(/, **data: typing.Any)#

Bases: pydantic.BaseModel

TLC patch request.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

delete_old_url: bool = False#

model_config = ConfigDict(...)#

class tlc.service.object_service.TLCPatchRequest(/, **data: typing.Any)#

Bases: pydantic.BaseModel

In-flight representation of a patch request for a TLCObject.

Create a new model by parsing and validating input data from keyword arguments.

Raises [ValidationError][pydantic_core.ValidationError] if the input data cannot be validated to form a valid model.

self is explicitly positional-only to allow self as a field name.

patch_object: tlc.service.object_service.TLCObject = None#

patch_options: tlc.service.object_service.TLCPatchOptions = TLCPatchOptions(...)#

model_config = ConfigDict(...)#

class tlc.service.object_service.RollbackDeleteContext(url: tlc.core.url.Url)#

A context manager for rollback object creation without interfering with InsufficientCredits.

rollback() → None#

tlc.service.object_service.get_ip_addresses() → list[str]#

tlc.service.object_service.get_running_urls() → list[str]#

tlc.service.object_service.format_yaml_for_logging(data: dict | list, indent: int = 4) → str#

tlc.service.object_service.open_in_web_browser(url: str) → None#

tlc.service.object_service.open_dashboard_in_web_browser(app: litestar.Litestar) → None#

async tlc.service.object_service.startup(app: litestar.Litestar) → None#: Setup HTTP client for connecting to 3LC Data Service

async tlc.service.object_service.shutdown(app: litestar.Litestar) → None#: Perform any required cleanup before terminating the application

async tlc.service.object_service.root() → litestar.Response#: Root endpoint of the service

class tlc.service.object_service.ObjectServiceFeatures(/, **data: typing.Any)#

Bases: pydantic.BaseModel

post_external_data: bool = True#

class tlc.service.object_service.ObjectServiceUserInfo(/, **data: typing.Any)#

Bases: pydantic.BaseModel

user_id: str = None#

tenant_id: str = None#

user_full_name: str = None#

user_email: str = None#

tenant_name: str = None#

class tlc.service.object_service.DashboardAnnotations(/, **data: typing.Any)#

Bases: pydantic.BaseModel

banner_icon_url: str = <Multiline-String>#

banner_background_color: str = <Multiline-String>#

banner_message: str = <Multiline-String>#

title_message: str = <Multiline-String>#

tlc.service.object_service.get_status() → dict[str, Any]#: Returns status of the service

tlc.service.object_service.last_lru_stats: dict[str, Any] | None = None#

tlc.service.object_service.get_last_lru_stats() → dict[str, Any] | None#

async tlc.service.object_service.status(request: litestar.Request) → Dict[str, Any]#: Returns status of the service

class tlc.service.object_service.ETagMiddleware(app: litestar.types.ASGIApp, **kwargs: Any)#

Bases: litestar.middleware.MiddlewareProtocol

A middleware that may serve 304 not modified responses based on ETag headers.

Only affects endpoints/responses that have previously been served with ETag. Other requests are passed through.

If the request contains an If-None-Match header with the same ETag as the previous response, a 304 Not Modified response is returned.

class tlc.service.object_service.DashboardKeyGuard(app: litestar.types.ASGIApp, exclude: str | list[str] | None = None, exclude_from_auth_key: str = 'exclude_from_auth', exclude_http_methods: Sequence[litestar.types.Method] | None = None, scopes: litestar.types.Scopes | None = None)#

Bases: litestar.middleware.authentication.AbstractAuthenticationMiddleware

async authenticate_request(connection: litestar.connection.ASGIConnection) → litestar.middleware.authentication.AuthenticationResult#

class tlc.service.object_service.ObjectRoutesController(owner: litestar.router.Router)#

Bases: litestar.controller.Controller

Controller for all object-related routes

Initialize a controller.

Should only be called by routers as part of controller registration.

Args: owner: An instance of :class:Router <.router.Router>

path = /objects#

cache_time_out = resolve_cache_timeout(...)#

async get_encoded_url(encoded_url: str, request: litestar.Request) → litestar.Response[tlc.service.object_service.TLCObject]#

async get_encoded_url_rows(encoded_url: str, attribute: str, request: litestar.Request) → litestar.Response[bytes]#

async list_urls() → List[str]#

Return all the objects.

Returns: List[Any]: List of the URLs of all the objects.

async new_object(data: tlc.service.object_service.TLCObject) → litestar.Response#

Create a new object.

Parameters:: data – Object to be created
Returns:: Empty response. URL of the created object will be in the ‘Location’ field of the response headers.

async delete_object(encoded_url: str) → None#

Delete an object.

Parameters:: encoded_url – URL of the object to be deleted.
Raises:: HTTPException if no object can be found at the URL.

async update_object(encoded_url: str, data: tlc.service.object_service.TLCPatchRequest) → litestar.Response#

Update the attributes of an object.

Raises: HTTPException: If the object type of obj_in does not match the type of the object at object_url.

class tlc.service.object_service.ExternalDataRoutesController(owner: litestar.router.Router)#

Bases: litestar.controller.Controller

Controller for all external data-related routes

Initialize a controller.

Should only be called by routers as part of controller registration.

Args: owner: An instance of :class:Router <.router.Router>

path = /external_data#

cache_time_out = resolve_cache_timeout(...)#

async get_encoded_url(encoded_url: str) → bytes#

async post_encoded_url(request: litestar.Request, owner_url: str, base_name: str, extension: str) → litestar.Response#

Write a new file with given binary contents.

Parameters:

request – The request object.
owner_url – The URL of the tlc Object that owns the file. Currently only support Table owners, data will be written in the table’s bulk data folder.
base_name – The base name of the file or folder to be created. Used to provide more context to the filename.
extension – The extension of the file.

Returns:

A response with a 201 status code and a Location header pointing to the newly created file.

async get_encoded_url_binary_contents(encoded_url: str, format: str) → litestar.Response#

class tlc.service.object_service.TLCCustomLoggingMiddleware(app: litestar.types.ASGIApp, config: litestar.middleware.logging.LoggingMiddlewareConfig)#

Bases: litestar.middleware.logging.LoggingMiddleware

Custom middleware to log object service requests and responses.

Logs request and response data to loglevel.INFO, together with the time it takes to complete the request.

Initialize LoggingMiddleware.

Args: app: The next ASGI app to call. config: An instance of LoggingMiddlewareConfig.

async log_request(scope: litestar.types.Scope, receive: litestar.types.Receive) → None#: Record the start time and log the request data.

log_response(scope: litestar.types.Scope) → None#: Measure elapsed time and log the response data.

log_message(values: dict[str, Any]) → None#

Log a message.

This is a copy of the superclass’ method, with special case handling of the /status endpoint, and url decoding of the path.

Parameters:: values – Extract values to log.
Returns:: None

class tlc.service.object_service.NGrokOutputAdaptor(role: str, ngrok_helper: tlccli.subcommands.ngrok_helper.NGrokHelper)#

Helper class to format output from NGrokHelper for the Object Service.

async output_public_url(app: litestar.Litestar) → None#

tlc.service.object_service.T = TypeVar(...)#

tlc.service.object_service.create_litestar_app(host: str, port: int, use_ngrok: bool, dashboard: bool = False, after_startup_handler: list[litestar.types.callable_types.LifespanHook] | litestar.types.callable_types.LifespanHook | None = None, after_shutdown_handler: list[litestar.types.callable_types.LifespanHook] | litestar.types.callable_types.LifespanHook | None = None) → litestar.Litestar#

tlc.service.object_service#

Module Contents#

Classes#

Functions#

Data#

API#

`tlc.service.object_service`#