Datapool

A funcionalidade do Datapool pode ser utilizada como uma forma de gerenciar uma fila de itens que precisam ser processados em lote.

Nessa seção você verá mais detalhes sobre como interagir com esse recurso do BotCity Maestro através do código da sua automação.

Dica

Veja mais detalhes sobre como criar um Datapool e adicionar novos itens através da interface do Maestro acessando este link.

Consumindo os itens de um Datapool

O primeiro passo é obtermos a referência do Datapool através do seu identificador único (label).

Após obtermos a referência do Datapool, podemos utilizar um laço de repetição para verificar enquanto houverem itens para serem processados.

PythonC#

# Consumindo o próximo item disponível e reportando o estado de finalização ao final
datapool = maestro.get_datapool(label="Items-To-Process")

while datapool.has_next():
    # Busca o próximo item do Datapool
    item = datapool.next(task_id=execution.task_id)
    if item is None:
        # O item poderia ser 'None', caso outro processo o consumisse antes
        break

    try:
        # Processando o item...

        # Finalizando como 'CONCLUÍDO' após o processamento
        item.report_done(finish_message="Processado com sucesso!")

    except Exception:
        # Finalizando o processamento do item como 'ERRO'
        item.report_error(finish_message="Falha no processamento.")

// Consumindo o próximo item disponível e reportando o estado de finalização ao final
Datapool datapool = await maestro.GetDatapoolAsync("Items-To-Process");

while (await dataPool.HasNextAsync()) {
    // Busca o próximo item do Datapool
    DatapoolEntry item = await datapool.NextAsync(execution.TaskId);
    if (item == null) {
        // O item poderia ser 'null', caso outro processo o consumisse antes
        break;
    }

    try {
        // Processando o item...

        // Finalizando como 'CONCLUÍDO' após o processamento
        await item.ReportDoneAsync();

    } catch (Exception ex) {
        // Finalizando o processamento do item como 'ERRO'
        await item.ReportErrorAsync();
    }
}

Aviso

Lembre-se de sempre incluir no código o reporte referente ao estado de finalização do item que foi processado.

Isso é extremamente importante para que os estados dos itens sejam atualizados dentro do Datapool no Maestro.

Manipulando itens do Datapool

Além de reportar o estado de finalização de um item, podemos realizar outras operações no código.

Acessando os dados

É possível obtermos algumas informações do item, além de dados específicos com base no Schema que foi criado.

PythonC#

# Busca o próximo item do Datapool
item = datapool.next(task_id=execution.task_id)

# Através do método get_entry() também podemos obter um item através do seu ID
item = datapool.get_entry(entry_id="<ENTRY_ID>")

# Obtendo o valor de algum campo específico do item
item_data = item["data-label"]

# Utilizar o método get_value() terá o mesmo efeito
item_data = item.get_value("data-label")

// Busca o próximo item do Datapool
DatapoolEntry item = await datapool.NextAsync(execution.TaskId);

// Obtendo o valor de algum campo específico do item
string item_data = await item.GetValueAsync("data-label");

Reportando o estado de finalização

Reportar o estado de finalização de um item é essencial para que eles sejam contabilizados da forma correta.

Cada item que é processado pode ser finalizado com um estado de CONCLUÍDO ou ERRO.

Em ambos os casos é possível passar uma mensagem de finalização. No caso de erros, também é possível especificar qual o tipo de erro.

PythonC#

# Busca o próximo item disponível do Datapool
item = datapool.next(task_id=execution.task_id)

# Finalizando como 'CONCLUÍDO' após o processamento
item.report_done(finish_message="Processado com sucesso!")

# Finalizando o processamento do item indicando um erro de sistema
item.report_error(error_type=ErrorType.SYSTEM, finish_message="Sistema indisponível.")

# Finalizando o processamento do item indicando um erro de negócio
item.report_error(error_type=ErrorType.BUSINESS, finish_message="Dados inválidos.")

Info

Por padrão, todo erro será considerado do tipo SYSTEM quando não for especificado no reporte.

Utilize a classe ErrorType para especificar o tipo de erro adequado.

// Busca o próximo item do Datapool
DatapoolEntry item = await datapool.NextAsync(execution.TaskId);

// Finalizando como 'CONCLUÍDO' após o processamento
await item.ReportDoneAsync();

// Finalizando o processamento do item como 'ERRO'
await item.ReportErrorAsync();

Operações com o Datapool

Através do Maestro SDK também podemos realizar algumas outras operações com o Datapool, além de verificar e consumir os itens a serem processados.

PythonC#

# Obtendo a referência do Datapool
datapool = maestro.get_datapool("ProductsData")

# Verificando se o Datapool está ativo
print(datapool.is_active())

# Verificando se o Datapool está vazio
print(datapool.is_empty())

# Marcando o Datapool como ativo
datapool.activate()

# Marcando o Datapool como inativo
datapool.deactivate()

// Obtendo a referência do Datapool
Datapool datapool = await maestro.GetDatapoolAsync("ProductsData");

// Verificando se o Datapool está vazio
Console.WriteLine(await datapool.IsActiveAsync());

// Checking if the Datapool is empty
Console.WriteLine(await datapool.IsEmptyAsync());

// Marcando o Datapool como ativo
await datapool.ActivateAsync();

// Marcando o Datapool como inativo
await datapool.DeactivateAsync();

Criando um Datapool via código

O Maestro SDK permite criarmos do zero a estrutura de um novo Datapool.

Além das configurações básicas, podemos definir as regras que serão utilizadas e também o Schema base dos itens.

Criação do Schema

Os campos que irão compor o Schema devem ser criados através da classe SchemaField.

Para cada campo, além do label e type, podemos definir se o campo deve possuir um valor único (unique_id) e também se ele deve ser exibido na listagem dos itens na fila (display_value).

Você também pode criar o Datapool passando um Schema vazio, caso não seja necessário utilizar uma estrutura base para os itens.

PythonC#

# Criando os campos que irão compor o Schema: 'id', 'name' e 'price'
# Os campos do Schema devem ser criados através da classe 'SchemaField'

product_id = SchemaField(
    label="id",             # Identificador único do campo
    type=FieldType.TEXT,    # Tipo do campo
    unique_id=True,         # O campo 'id' será um campo com valor único
    display_value=True      # Esse campo também será exibido na listagem dos itens
)

product_name = SchemaField(
    label="name",
    type=FieldType.TEXT,
    unique_id=False,
    display_value=True
)

product_price = SchemaField(
    label="price",
    type=FieldType.DOUBLE,
    unique_id=False,
    display_value=False
)

# Campos que serão utilizados no Schema do Datapool
schema = [product_id, product_name, product_price]

# Criando o objeto do Datapool
datapool = DataPool(
    label="ProductsData",   # Identificador único do Datapool
    name="ProductsData",    # Nome de exibição
    max_auto_retry=2,       # Número máximo de reprocessamentos em caso de erro
    max_errors_before_inactive=5,   # Número máximo de erros consecutivos permitidos
    item_max_processing_time=3,     # Tempo máximo de processamento de cada item (em minutos)
    schema=schema   # Campos que irão compor a estrutura de cada item ('schema' criado acima)
)

# Criando a estrutura do Datapool no Orquestrador
maestro.create_datapool(datapool)

// Ainda não implementado

Dica

Apesar do Maestro SDK possibilitar a criação de um Datapool via código, recomendamos que você utilize a interface web do Orquestrador para que consiga realizar as configurações de forma mais intuitiva. Mais detalhes aqui.

Uma vez que a estrutura do Datapool está criada na plataforma, você pode utilizar o Maestro SDK para a criação e consumo de novos itens.

Adicionando novos itens

Podemos adicionar novos itens ao Datapool utilizando o Maestro SDK.

Dica

Esse método pode ser útil caso você queira utilizar um script Python para popular o Datapool com os itens que serão processados posteriormente.

PythonC#

# Instanciando um novo item do Datapool com base no Schema que foi definido
new_item = DataPoolEntry(
    values={
        "id": "Electronic#001"
        "name": "Smartphone",
        "price": "2000"
    }
)

# Obtendo a referência do Datapool
datapool = maestro.get_datapool("ProductsData")

# Adicionando um novo item
datapool.create_entry(new_item)

// Instanciando um novo item do Datapool com base no Schema que foi definido
var values = new Dictionary<string, object>
{
    { "id", "Electronic#001" },
    { "name", "Smartphone" },
    { "price", 2000 }
};
DatapoolEntry new_item = new DatapoolEntry(0,values);

// Obtendo a referência do Datapool
Datapool datapool = await maestro.GetDatapoolAsync("ProductsData");

// Adicionando um novo item
await datapool.CreateEntryAsync(new_item);

Cancelando itens

Caso por algum motivo seja necessário ignorar um item específico que foi criado, podemos utilizar a funcionalidade de cancelar.

Desse modo, um item que foi cancelado não será puxado para processamento, mas ainda sim aparecerá no histórico do Datapool.

Info

Somente itens que possuem o estado de PENDENTE podem ser cancelados.

PythonC#

# Cancelando item que está pendente na fila
datapool.cancel_entry(
    entry_id="<ENTRY_ID>",
    finish_message="Item com dados faltantes"
)

// Ainda não implementado

Deletando itens

Caso você queira remover um item do Datapool, você pode utilizar a funcionalidade de deletar.

Nesse caso, itens pendentes serão excluídos da fila e itens já processados serão removidos do histórico.

Info

Itens que estão no estado de PROCESSANDO ou TIMEOUT não podem ser excluídos.

PythonC#

# Removendo item da fila
datapool.delete_entry(entry_id="<ENTRY_ID>")

// Ainda não implementado