fsglue 1.2.0

cloud firestore orm inspired by ndb

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

• API Reference