Skip to content
Open
Show file tree
Hide file tree
Changes from all commits
Commits
File filter

Filter by extension

Filter by extension

Conversations
Failed to load comments.
Loading
Jump to
Jump to file
Failed to load files.
Loading
Diff view
Diff view
134 changes: 132 additions & 2 deletions README.md
Original file line number Diff line number Diff line change
@@ -1,4 +1,134 @@
# sbxcloudpython
# SbxCloudPython

This is the first version of Python Library for SbxCloud. The min Python version require is 3.5, this library depends on courutines with asyncio in order to use concurrent task, and uses aiohttp to do the request. You can test using the test.py file, before execute it, configuring your credentials in the environment.
A Python client library for interacting with SbxCloud services.

## Overview

SbxCloudPython is a Python client library that provides a convenient way to interact with SbxCloud services. It supports both synchronous and asynchronous operations, making it flexible for various use cases.

## Requirements

- Python 3.5 or higher
- Dependencies:
- `asyncio` - For asynchronous operations
- `aiohttp` - For HTTP requests

## Installation

You can install the library using pip:

```bash
pip install sbxpy
```

## Features

- Query building with type hints for better IDE support
- Asynchronous operations with asyncio
- Callback-based API for synchronous code
- Support for complex queries with reference joins
- Pagination handling

## Basic Usage

### Initialization

```python
from sbxpy import SbxCore

# Initialize the SbxCore instance
sbx = SbxCore()
sbx.initialize(domain="your_domain", app_key="your_app_key", base_url="https://api.sbxcloud.com")

# For async operations
import asyncio

async def main():
# Login to SbxCloud
login_result = await sbx.login("your_email", "your_password", "your_domain")
print(login_result)

# Run the async function
asyncio.run(main())
```

### Building Queries

```python
# Create a query to find data
find = sbx.with_model("your_model")
find.and_where_is_equal("field_name", "value")
find.and_where_greater_than("numeric_field", 100)
find.set_page(1)
find.set_page_size(50)

# Execute the query asynchronously
async def execute_query():
results = await find.find()
print(results)

asyncio.run(execute_query())
```

### Using Callbacks

```python
# Initialize with event loop management
sbx = SbxCore(manage_loop=True)
sbx.initialize(domain="your_domain", app_key="your_app_key", base_url="https://api.sbxcloud.com")

# Define a callback function
def on_login_complete(error, data):
if error:
print(f"Error: {error}")
else:
print(f"Login successful: {data}")

# Use the callback API
sbx.loginCallback("your_email", "your_password", "your_domain", on_login_complete)

# Don't forget to close the connection when done
sbx.close_connection()
```

## Advanced Usage

### Reference Joins

```python
# Create a query with reference joins
find = sbx.with_model("your_model")
reference_join = find.and_where_reference_join_between("field_in_model", "field_in_referenced_model")
filter_join = reference_join.in_model("referenced_model")
filter_join.filter_where_is_equal("value")

# Execute the query
async def execute_complex_query():
results = await find.find()
print(results)

asyncio.run(execute_complex_query())
```

### Fetching Related Models

```python
find = sbx.with_model("your_model")
find.fetch_models(["related_model1", "related_model2"])
find.and_where_is_equal("field_name", "value")

async def fetch_with_related():
results = await find.find()
print(results)

asyncio.run(fetch_with_related())
```

## Testing

You can test the library using the provided `test.py` file. Before executing it, make sure to configure your credentials in the environment.

## License

This project is licensed under the terms of the license included in the repository.

38 changes: 20 additions & 18 deletions sbxpy/QueryBuilder.py
Original file line number Diff line number Diff line change
@@ -1,70 +1,72 @@
from typing import Dict, List, Any, Union, Optional

class QueryBuilder:
'''
.. class:: QueryBuilder
:platform: Unix, Windows
:synopsis: This is the main class uses to create the json params in order to do a find and delete request.
.. moduleauthor:: Luis Guzman <lgguzman890414@gmail.com>
'''
def __init__(self):
self.q = {"page": 1, "size": 1000, "where": []}
self.group = {
def __init__(self) -> None:
self.q: Dict[str, Any] = {"page": 1, "size": 1000, "where": []}
self.group: Dict[str, Any] = {
"ANDOR": "AND",
"GROUP": []
}
self.OP = ["in", "IN", "not in", "NOT IN", "is", "IS", "is not", "IS NOT", "<>", "!=", "=", "<", "<=", ">=", ">",
self.OP: List[str] = ["in", "IN", "not in", "NOT IN", "is", "IS", "is not", "IS NOT", "<>", "!=", "=", "<", "<=", ">=", ">",
"like", "LIKE", "not like", "NOT LIKE"]

def set_domain(self, domain_id):
def set_domain(self, domain_id: str) -> 'QueryBuilder':
self.q['domain'] = domain_id
return self

def set_model(self, model_name):
def set_model(self, model_name: str) -> 'QueryBuilder':
self.q['row_model'] = model_name
return self

def set_page(self, page):
def set_page(self, page: int) -> 'QueryBuilder':
self.q['page'] = page
return self

def set_page_size(self, page_size):
def set_page_size(self, page_size: int) -> 'QueryBuilder':
self.q['size'] = page_size
return self

def order_by(self, field, asc):
def order_by(self, field: str, asc: Optional[bool] = None) -> 'QueryBuilder':
if asc is None:
asc = False
self.q['order_by'] = {'ASC': asc, 'FIELD': field}
return self

def fetch_models(self, array_of_model_names):
def fetch_models(self, array_of_model_names: List[str]) -> 'QueryBuilder':
self.q['fetch'] = array_of_model_names
return self

def reverse_fetch(self, array_of_model_names):
def reverse_fetch(self, array_of_model_names: List[str]) -> 'QueryBuilder':
self.q['rev_fetch'] = array_of_model_names
return self

def add_object_array(self, array):
def add_object_array(self, array: List[Dict[str, Any]]) -> 'QueryBuilder':
if 'where' in self.q:
del self.q['where']
if 'rows' not in self.q:
self.q['rows'] = []
self.q['rows'] = self.q['rows'] + array
return self

def add_object(self, object):
def add_object(self, object: Dict[str, Any]) -> 'QueryBuilder':
if 'where' in self.q:
del self.q['where']
if 'rows' not in self.q:
self.q['rows'] = []
self.q['rows'].append(object)
return self

def where_with_keys(self, keys_array):
def where_with_keys(self, keys_array: List[str]) -> 'QueryBuilder':
self.q['where'] = {"keys": keys_array}
return self

def new_group(self, connector_AND_or_OR):
def new_group(self, connector_AND_or_OR: str) -> 'QueryBuilder':
if 'rows' in self.q:
del self.q['rows']
if len(self.group['GROUP']) > 0:
Expand All @@ -75,7 +77,7 @@ def new_group(self, connector_AND_or_OR):
}
return self

def set_reference_join(self, operator, filter_field, reference_field, model, value):
def set_reference_join(self, operator: str, filter_field: str, reference_field: str, model: str, value: Any) -> 'QueryBuilder':
self.q["reference_join"] = {
"row_model": model,
"filter": {
Expand All @@ -87,7 +89,7 @@ def set_reference_join(self, operator, filter_field, reference_field, model, val
}
return self

def add_condition(self, connector_AND_or_OR, field_name, operator, value):
def add_condition(self, connector_AND_or_OR: str, field_name: str, operator: str, value: Any) -> 'QueryBuilder':
if len(self.group['GROUP']) < 1:
connector_AND_or_OR = "AND"
self.group['GROUP'].append({
Expand All @@ -99,7 +101,7 @@ def add_condition(self, connector_AND_or_OR, field_name, operator, value):

return self

def compile(self):
def compile(self) -> Dict[str, Any]:
if 'where' in self.q:
if 'rows' in self.q:
del self.q['rows']
Expand Down
Loading