Hey guys, I'm just starting using prefect cloud. I...
# ask-communitya
ARun
09/03/2020, 12:34 PMHey guys, I'm just starting using prefect cloud. I have the prefect agent up and running registered with the cloud. I can see the agent alive on the prefect cloud gui as well. However, when I try to register my first flow, I get the error
Copy code
ClientError: Malformed response received from Cloud - please ensure that you have an API token properly configured.j
Jenny
09/03/2020, 1:15 PMHi @ARun - For Cloud you'll need a Personal Access Token to login and a runner token for your agent. You can see more about getting started here: https://docs.prefect.io/orchestration/tutorial/configure.html#select-an-orchestration-backend
a
ARun
09/03/2020, 1:26 PMI registered the client, but not I'm getting this error
Copy code
ClientError: [{'path': ['project'], 'message': 'field "project" not found in type: \'query_root\'', 'extensions': {'path': '$.selectionSet.project', 'code': 'validation-failed', 'exception': {'message': 'field "project" not found in type: \'query_root\''}}}]ARun
09/03/2020, 1:26 PMI can verify that that project has been created
j
Jenny
09/03/2020, 1:28 PMHave you run
prefect auth login -t PERSONALACCESSTOKENa
ARun
09/03/2020, 3:04 PM@Jenny Thanks, that did it. Another question with the flow. I made a simple task to run a sql command.
then set up the flow to run the task.
the flow runs fine without any issues, but when i try to register it, I get an error
Copy code
TypeError: cannot pickle '_thread.lock' objectARun
09/03/2020, 3:04 PMmy code
Copy code
def runSql(sql, name=None):
cs.execute(sql)
with Flow('run') as flow:
runSql("USE ROLE SYSADMIN")
flow.register(project_name="test")ARun
09/03/2020, 3:05 PMfull error
Copy code
c:\python38\lib\site-packages\prefect\core\flow.py in register(self, project_name, build, labels, set_schedule_active, version_group_id, no_url, **kwargs)
1586 client = prefect.Client()
1587
-> 1588 registered_flow = client.register(
1589 flow=self,
1590 build=build,
c:\python38\lib\site-packages\prefect\client\client.py in register(self, flow, project_name, build, set_schedule_active, version_group_id, compressed, no_url)
732 )
733
--> 734 serialized_flow = flow.serialize(build=build) # type: Any
735
736 # Set Docker storage image in environment metadata if provided
c:\python38\lib\site-packages\prefect\core\flow.py in serialize(self, build)
1429 raise ValueError("This flow has no storage to build")
1430 if self.name not in self.storage:
-> 1431 self.storage.add_flow(self)
1432 else:
1433 warnings.warn(
c:\python38\lib\site-packages\prefect\environments\storage\local.py in add_flow(self, flow)
131 self.directory, "{}.prefect".format(slugify(flow.name))
132 )
--> 133 flow_location = flow.save(flow_location)
134
135 self.flows[flow.name] = flow_location
c:\python38\lib\site-packages\prefect\core\flow.py in save(self, fpath)
1498 fpath.parent.mkdir(exist_ok=True, parents=True)
1499 with open(str(fpath), "wb") as f:
-> 1500 cloudpickle.dump(self, f)
1501
1502 return str(fpath)
c:\python38\lib\site-packages\cloudpickle\cloudpickle_fast.py in dump(obj, file, protocol, buffer_callback)
53 compatibility with older versions of Python.
54 """
---> 55 CloudPickler(
56 file, protocol=protocol, buffer_callback=buffer_callback
57 ).dump(obj)
c:\python38\lib\site-packages\cloudpickle\cloudpickle_fast.py in dump(self, obj)
561 def dump(self, obj):
562 try:
--> 563 return Pickler.dump(self, obj)
564 except RuntimeError as e:
565 if "recursion" in e.args[0]:
TypeError: cannot pickle '_thread.lock' objectj
Jenny
09/03/2020, 3:45 PMHi @ARun - I've got no immediate answer about what's causing that for you. I found this to be a good explanation of what that error message means. It may be from cs.execute? What is cs? Also, you might find one of our task library tasks can help you with SQL or give you a good example of something you could work from to develop your own: https://docs.prefect.io/core/task_library/