Hi Everyone, I'm new to prefect and I'm running i...
# ask-community
r
Hi Everyone, I'm new to prefect and I'm running into the following issue. I defined a task and inside a flow block I call this task. It happens that I expect the return of this task to be a list of urls(variable "extratos"), but don't know why I'm getting the following error message when "len(arquivos_extrato" is run: "object of type 'GetItem' has no len()". sharepoint4 is a customized class that i have built and the function listar_arquivos() is part of this class. My code is something like this: Can someone help me to understand why am I getting a GetItem object and not a list ?
1
👋 2
c
Howdy @Rafael Sá!
k
Hi @Rafael Sá, welcome! You are probably doing something like:
Copy code
@task
def abc():
    return [1,2,3]

with Flow(..) as flow:
    x = abc()
    len(x)
you can’t do this because the tasks defer execution to
flow.run()
or when the flow is run with an agent so
x
is still of type
Task
here. Try only having tasks inside the Flow block so the
len()
operation should be in a task
r
Hi @Kevin Kho, thanks for the feedback, thats exactly what I'm doing but really got confused since I have all other tasks in the flow running like that. For example I call anothe task inside the Flow block thtat is:
Copy code
destino_insumos,caminho_destino_planilhas_fluxos,destino_bank1,destino_bank2 =configurar_caminho_diretorios(ano_referencia,mes_referencia,mes_bbank1,mes_bank2)
all those variables are normally assigned to the correct type.
k
It will be after the flow run but you can’t do something like
destino_insumos.attribute
inside the Flow because it’s not that type yet.
Like if this is a pandas DataFrame, you can’t call
.shape
destino_insumos,caminho_destino_planilhas_fluxos,destino_bank1,destino_bank2
are all of type Task while the Flow is constructed
r
So I can only pass the assigned variable to a next task, but can't operate directly over it inside the flow block ? That is it?
k
Yeah exactly so pretty much all the logic inside the Flow block should be tasks
r
great @Kevin Kho. thanks so much.
k
Oh can we move the code to the thread to keep the main channel cleaner when you get the chance?
r
Sorry, how can i do that ?
k
Just edit the first post and copy the code and paste it here
r
Copy code
from sharepoint4 import *
import os
from datetime import datetime
import calendar
from datetime import timedelta
import locale
from shutil import copy
from dateutil.relativedelta import relativedelta
import openpyxl
import prefect
from prefect.run_configs import LocalRun
from prefect import task,Flow


@task(nout=2)
def obter_lista_origem_destino(meu_site,SP_DOC_LIBRARY,filtro,split_string,base_path_destino):

    #configura o endereço de origem do arquivo e o destino caso deseje baixar o arquivo
    origem = ""
    destino = ""

    #instancia objeto da classe customizada sharepoint
    meu_sharepoint = Sharepoint(meu_site,origem,destino)

    #constroi a lista de arquivos desejados a serem copiados
    extratos = meu_sharepoint.listar_arquivos(SP_DOC_LIBRARY,filtro)

    #configura o caminho do destino dos arquivos que irão do sharepoint para o servidor local      considerando o caminho de origem
    destino_fluxo = [destino.split(split_string)[1] for destino in extratos]

    #adiciona o drive para a gravação do arquivo
    servidor = [(base_path_destino+(destino)) for destino in destino_fluxo]
    return extratos,servidor

#início do código principal(definição de variáveis globais)
with Flow("CRIA DIRETORIOS") as flow:

  #define o site do sharepoint no qual se autenticar para acessar a biblioteca de documentos
  meu_site = "<https://xxx/sites/yyy>"

  #define a biblioteca de documentos a ser pesquisada
  SP_DOC_LIBRARY ="Documentos"


arquivos_extrato,destino_servidor = obter_lista_origem_destino(meu_site,SP_DOC_LIBRARY,filtro_dir,split_string,base_path_destino)

for i in range(0,len(arquivos_extrato)):
    copiar_insumo(arquivos_extrato[i],destino_servidor[i])
k
Thank you!