Skip to main content

cloud firestore orm inspired by ndb

Project description

fsglue

Simple ORM for google cloud firestore inspired by datastore client library ndb and used by Bizglue(Japanese Only).

Installation

pip install fsglue

Simple Usage

import fsglue

class Fruit(fsglue.BaseModel):
    COLLECTION_PATH = "fruit"
    COLLECTION_PATH_PARAMS = []

    name = fsglue.StringProperty(required=True)
    price = fsglue.IntegerProperty(required=True)


## create
apple = Fruit.create()
apple.name = "apple"
apple.price = 100
apple.put()

# get
apple = Fruit.get_by_id(apple.doc_id)
apple = Fruit.where([["name", "==", "apple"]])[0]

# update
apple.price = 120
apple.put()

# delete
apple.delete()

Client Initialization

case1

Initialize from the environment. No code will be needed.

case2

Initialize by firestore.Client. Following code will pass kwargs to firestore.Client(**kwargs).

fsglue.initialize(**kwargs)

case3

Initialize by firebase_admin.

import firebase_admin
from firebase_admin import credentials
from firebase_admin import firestore
cred = credentials.Certificate('path/to/serviceAccount.json')
firebase_admin.initialize_app(cred)

Model Samples

class User(fsglue.BaseModel):
    COLLECTION_PATH = "users"
    COLLECTION_PATH_PARAMS = []

    name = fsglue.StringProperty(required=True)
    created_at = fsglue.TimestampProperty(auto_now=True)
    updated_at = fsglue.TimestampProperty(auto_now_add=True)

    @classmethod
    def create_by_name(cls, name):
        return cls.create_by_dict({"name": name})


# JsonSchema Definition for JsonProperty
TAGS_SCHEMA = {
    "type": "array",
    "items": {
        "type": "string",
    },
}


class Room(fsglue.BaseModel):
    COLLECTION_PATH = "rooms"
    COLLECTION_PATH_PARAMS = []

    name = fsglue.StringProperty(required=True)
    owner = fsglue.StringProperty(required=True)
    tags = fsglue.JsonProperty(schema=TAGS_SCHEMA, default=[])
    created_at = fsglue.TimestampProperty(auto_now=True)
    updated_at = fsglue.TimestampProperty(auto_now_add=True)

    @classmethod
    def create_by_user(cls, user, tags=[]):
        room = cls.create()
        room.name = "untitled"
        room.owner = user.doc_id
        room.tags = tags
        room.put()
        return room

    def post_message(self, user, body):
        msg = Message.create(self.doc_id)
        msg.body = body
        msg.created_by = user.doc_id
        msg.put()
        return msg

    def fetch_latest_messages(self):
        return Message.all(self.doc_id, limit=100, order_by="-created_at")


class Message(fsglue.BaseModel):
    COLLECTION_PATH = "rooms/{0}/messages"
    COLLECTION_PATH_PARAMS = ["room_id"]

    body = fsglue.StringProperty(required=True)
    created_by = fsglue.StringProperty(required=True)
    created_at = fsglue.TimestampProperty(auto_now=True)

Operation Samples

Create

Using .create() and .put()

john = User.create()
john.name = "john"
john.put()

room1 = Room.create()
room1.name = "test"
room1.owner = john.doc_id
room1.tags = ["test"]
room1.put()

message11 = Message.create(room1.doc_id)  # parent_id = room1.doc_id
message11.body = "hello fsglue!"
message11.put()

Using .create_by_dict()

john = User.create_by_dict({"name": "john"})
room1 = Roomt.create_by_dict({"name": "test", "owner": john.doc_id, "tags": ["test"]})
message11 = Message.create_by_dict({"body": "hello fsglue!"}, room1.doc_id) # parent_id = room1.doc_d

Using custom method

john = User.create_by_name("john")
room1 = Room.create_by_user(john, tags=["test"])
message11 = Room.post_message(john, "hello fsglue!")

Get

Using .get_by_id()

john = User.get_by_id("xxx")  # return None if not exists
room1 = Room.get_by_id("yyy")
message11 = Message.get_by_id("zzz", room1.doc_id)

Using .get_by_ids()

john = User.get_by_ids(["xxx"])[0]
room1 = Room.get_by_ids(["yyy"])[0]
message11 = Message.get_by_ids(["zzz"], room1.doc_id)[0]

Using .all()

users = User.all(limit=10)
rooms = Room.all(limit=10)
messages1 = Message.all(room1.doc_id, limit=10)  # get messages belong to room1

Using .where()

john = User.where([["name", "==", "john"]])[0]
room1 = Room.where([["name", "==", "room1"]])[0]
messages11 = Message.where([["body", "==", "hello fsglue!"]])[0]

Using .stream()

for user in User.stream():  # iterate all users
    print(user)

Update

Using .get_by_id() and .put()

john = User.get_by_id("xxx")
john.name = "john!"
john.put()

Using .update_by_dict()

john = User.update_by_dict({"id": "xxx", "name": "john!"})  # values require DocumentId in "id" field
room1 = Room.update_by_dict({"id": "yyy", "name": "test1"})
message11 = Message.update_by_dict({"id": "zzz", "body": "hello fsglue!?"}, room1.doc_id)

Delete

Using .get_by_id() and .delete()

john = User.get_by_id("xxx")
john.delete()

Using .get_by_id() and .delete_all()

room1 = Room.get_by_id("yyy")
room1.delete_all()  # delete_all() will delete room1 and messages belong to room1

Generate JsonSchema

Room.to_schema() will generate following JsonSchema definition

{
  "type": "object",
  "required": [
    "name",
    "owner"
  ],
  "properties": {
    "id": {
      "type": "string"
    },
    "name": {
      "type": "string"
    },
    "tags": {
      "type": "array",
      "items": {
        "type": "string"
      }
    },
    "owner": {
      "type": "string"
    },
    "created_at": {
      "type": "number"
    },
    "updated_at": {
      "type": "number"
    }
  }
}

Links

Project details


Download files

Download the file for your platform. If you're not sure which to choose, learn more about installing packages.

Source Distribution

fsglue-1.2.0.tar.gz (12.6 kB view hashes)

Uploaded Source

Built Distribution

fsglue-1.2.0-py3-none-any.whl (12.2 kB view hashes)

Uploaded Python 3

Supported by

AWS AWS Cloud computing and Security Sponsor Datadog Datadog Monitoring Fastly Fastly CDN Google Google Download Analytics Microsoft Microsoft PSF Sponsor Pingdom Pingdom Monitoring Sentry Sentry Error logging StatusPage StatusPage Status page