Community

Table of Contents

Installation

pip install git+https://github.com/jnsougata/deta.py

Quick Start

importdetaimportasyncioasyncdefmain():service=deta.Deta(DETA_PROJECT_KEY)# instantiating a basebase=service.base(name='TEST_BASE')# instantiating a drivedrive=service.drive(name='TEST_DRIVE')# put single json deta baseawaitbase.put(deta.Record({'name':'John Doe','age':20},key='xyz',expire_after=100)  )# or put multiple records with a single requestawaitbase.put(deta.Record({'name':'John Doe 0','age':20},key='xyz_1',expire_after=100),deta.Record({'name':'John Doe 1','age':21},key='xyz_2',expire_after=100),deta.Record({'name':'John Doe 2','age':22},key='xyz_3',expire_after=100)  )# doing queriesq=deta.Query()q.equals("name","John")q.equals("address.city","New York")q.range("age",18,30)q.not_equals("position","Manager")q.contains("id","-")results=awaitbase.fetch([q])print(results)# updating recordsu=deta.Updater()u.set("inactive",True)u.increment("age")u.delete("address")awaitbase.update("user_777",u)# downloading a song stored in deta drivestream=awaitdrive.get('song.mp3')asyncforchunk,_instream.iter_chunks():# do something with the chunk    ...# or read the entire filecontent=awaitstream.read()# do something with the content# closing deta connectionawaitservice.close()if__name__=='__main__':asyncio.run(main())

Async Context Manager

importdetaasyncdefmain():asyncwithdeta.Deta(DETA_PROJECT_KEY)asservice:base=service.base('TEST_BASE')print(awaitbase.get())drive=service.drive('TEST_DRIVE')stream=awaitdrive.get('song.mp3')withopen('song.mp3','wb')asf:f.write(awaitstream.read())

Documentation