| Name | suplex JSON |
| Version |
0.2.5
JSON |
| download |
| home_page | None |
| Summary | A Python library for interacting with Supabase REST API in Reflex applications |
| upload_time | 2025-07-21 19:53:49 |
| maintainer | None |
| docs_url | None |
| author | None |
| requires_python | >=3.12 |
| license | MIT |
| keywords |
authentication
database
reflex
supabase
|
| VCS |
 |
| bugtrack_url |
|
| requirements |
No requirements were recorded.
|
| Travis-CI |
No Travis.
|
| coveralls test coverage |
No coveralls.
|
# Suplex
Simple state module to manage user auth and create database queries with the Reflex web framework.
---
## Install
Add Suplex to your project.
```bash
# Install uv with curl
curl -LsSf https://astral.sh/uv/install.sh | sh
# Install uv with wget
wget -qO- https://astral.sh/uv/install.sh | sh
# Activate your venv in your base folder
cd /path/to/project
uv venv # or uv venv my-project-venv
source .venv/bin/activate # or source /my-project-venv/bin/activate
# Add suplex as dependency
uv add suplex
```
---
## Configure
In your project top-level directory, where rxconfig.py is located create a .env file as follows...
```bash
api_url="your-api-url"
api_key="your-api-key"
jwt_secret="your-jwt-secret"
service_role="your-service-role"
```
These values can be retrieved from Supabase. Log In >> Choose Project >> Project Settings >> Data API
Then in rxconfig.py add...
```python
from dotenv import load_dotenv
load_dotenv()
api_key = os.getenv("api_key")
api_url = os.getenv("api_url")
jwt_secret = os.getenv("jwt_secret")
# service_role = os.getenv("service_role") Only for admin use.
config = rx.Config(
# You may have a few entries here...
suplex={
"api_url": api_url,
"api_key": api_key,
"jwt_secret": jwt_secret
"let_jwt_expire": False # (Optional: Default is False) Specify if tokens auto refresh. Can set to True for tighter/manual control of token refresh
"cookie_max_age": 3600 # (Optional: Default = None) Seconds until cookie expires, otherwise is a session cookie.
}
)
```
---
## Build into existing State Class
Import Suplex, and subclass the module at the lowest layer. A BaseState class that is already built, or fresh will work. For a BaseState that is already built, make sure that you check the suplex functions and vars to ensure there aren't any collisions.
```python
from suplex import Suplex
class BaseState(Suplex):
# Your class below...
```
---
## Other Subclassing
For any other classes within your Reflex project, subclass your BaseState to give them access to the auth information and query methods. There shouldn't be any classes in your state requiring auth that don't inherit from the BaseState.
```python
class OtherState(BaseState):
# Your class below...
```
---
## Auth
Suplex comes with built-in vars and functions to manage users, user objects, JWT claims and more. Because front-end handling is different from project to project, you'll need to create functions for how you'd like to update your UI, redirect on success/failure, and handle exceptions.
### Auth Functions
- sign_up(email: str, phone: str, password: str, options: Dict[str, Any])
- Sign a user up using email or phone, and password. May pass in options - email_redirect_to, data, captcha_token, and channel.
- sign_in_with_password(email: str, phone: str, password: str, options: Dict[str, Any])
- Sign a user in with email or phone, and password. May pass in options - captcha_token.
- Returns user object dict and other info.
- sign_in_with_oauth(provider: str, options: Dict[str, Any])
- Returns a url to use with rx.redirect to send user to OAuth endpoint. May pass in options redirect_to, scopes, query_params, code_challenge, and code_challenge_method. Will need to manually parse the url that OAuth redirects to for storing the tokens. Use set_tokens() to set access_token and refresh_token.
- exchange_code_for_session(params: Dict[str, Any])
- Provide params auth_code, code_verifier for an access_token and refresh_token when using PKCE flow. Sets tokens automatically. Returns the user object and other information.
- set_tokens(access_token: str, refresh_token: str)
- Sets both tokens as well as setting the bearer token of the query object using the access_token. Use this if manually parsing a URL to set tokens.
- reset_password_email(email: str)
- Send password reset email to specified email address.
- get_user()
- Returns the user object of the current authenticated user from Supabase.
- update_user(email: str, phone: str, password: str, user_metadata: Dict[str, Any])
- Updates the current users email, phone, password, or custom data stored in user_metadata.
- Returns the updated user object.
- refresh_session()
- Manually refreshes the authentication session using the stored refresh token.
- get_settings()
- Returns the authentication settings for the Supabase project.
- log_out()
- Clears local tokens and attempts to nullify the refresh_token to Supabase.
- session_manager(event: Callable, on_failure: List[Callable] | Callable | None)
- Pass an event. If users token has expired, will attempt to refresh the token. If the token is unable to be refreshed, will trigger the on_failure event. Used typically to redirect to login page on failure.
Check docstrings for params, returns and exceptions.
### Exceptions
Auth requests are handled using the httpx library. Suplex raises no specific exceptions, so all issues with connection will be raised via the httpx exception classes. You can specify handling specific status codes as well (see examples below).
[QuickStart - HTTPX Exceptions](https://www.python-httpx.org/quickstart/#exceptions)
[HTTP response status codes - HTTP | MDN](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Status)
```python
from suplex import Suplex
import typing
class BaseState(Suplex):
# Login example.
def log_in(self, email, password) -> typing.Generator:
try:
self.sign_in_with_password(email, password)
except httpx.HTTPStatusError as exc:
if exc.response.status_code == 401:
yield rx.toast.error("Invalid email or password.")
except Exception:
yield rx.toast.error("Login failed.")
# Update user example.
def update_user_info(self, email, phone, password, user_metadata) -> typing.Generator:
try:
self.update_user(email, phone, password, user_metadata)
except httpx.HTTPStatusError as exc:
if exc.response.status_code == 403:
# Refresh token here and try again.
except Exception:
yield rx.toast.error("Updating user info failed.")
def log_out(self):
try:
self.logout()
except httpx.HTTPStatusError:
yield rx.toast.error("Unable to logout remotely, local session cleared.")
except Exception:
yield rx.toast.error("Something went wrong during logout.")
```
### Auth Vars
There is a full set of vars that pull values from the signed JWT that gets provided from Supabase in the form of an access_token. These vars pull those claims. If you don't want to use local information and instead only want to rely on a user object pulled directly from Supabase then you will want to use the get_user() function and parse the user object directly.
- user_id
- str - user's id as uuid.
- user_email
- str - user's email.
- user_phone
- str = user's phone.
- user_audience
- str - audience that user is a part of. Currently only supports audience "authenticated" as the JWT checks for this audience for validation.
- user_role
- str - role assigned to user.
- claims_issuer
- str - url that issued claim.
- claims_expire_at
- int - unix timestamp for when token expires.
- claims_issued_at
- int - unix timestamp for when token was issued.
- claims_session_id
- str - unique session id from Supabase.
- user_metadata
- Dict[str, Any] - custom data within user object.
- app_metadata
- Dict[str, Any] - extra info stored by Supabase.
- user_aal
- Literal["aal1" | "aal2"] - specifies 1 or 2 factor auth was used.
- user_is_authenticated
- bool - if user audience is "authenticated" will be True.
- user_is_anonymous
- bool - if user is anonymous will be True.
- user_token_expired
- bool - checks expiry for claims.
```python
# Frontend
def auth_component() -> rx.Component:
# Show only if user is logged in.
return rx.cond(
BaseState.user_is_authenticated,
shown_if_authenticated(),
shown_if_not_authenticated()
)
def user_info_panel() -> rx.Component:
# Show currently logged in user info.
return rx.flex(
rx.text(BaseState.user_id),
rx.text(BaseState.user_email),
rx.text(BaseState.user_phone),
class_name="flex-col items-center w-full"
)
# Setup a page to use auth_flow. Redirects user who isn't logged in.
@rx.page("/recipes", on_load=BaseState.auth_flow)
def recipes() -> rx.Component:
return rx.flex(
rx.button("Get Recipes")
on_click=BaseState.get_recipes()
)
class BaseState(Suplex):
def auth_flow(self) -> Callable:
if not self.user_is_authenticated:
return rx.redirect("/login")
```
---
## Query
Queries are built using the query() function to return an instance of the query that is then chained with other functions and then executed. Each query is it's own instance that executes in order for multiple queries to not overwrite data within Reflex's asynchronous event handler.
The style of query is similar to the official Supabase python client located at - [Python API Reference | Supabase Docs](https://supabase.com/docs/reference/python/select).
If a user's bearer token has expired, Suplex handles refreshing the session as long as within the initial rxconfig.py setup, let_jwt_expire is set to False (the default). If you prefer to manually handle jwt_expiration, or the refresh_session fails to refresh user's tokens properly, the query() function will raise a BearerTokenExpired exception.
```python
from suplex import Suplex, BearerTokenExpired
import typing
class BaseState(Suplex):
ingredients: list
recipes: list
def get_all_ingredients(self) -> typing.Generator:
try:
ingredients = []
# From 'recipes' table, select row ingredients
results = self.query().table("recipes").select("ingredients").execute()
for result in results:
ingredients.extend(result["ingredients"])
# Ensure no duplicate ingredients are present
ingredients = list(set(ingredients)
# Set ingredients to State
yield BaseState.setvar("ingredients", ingredients)
# If users token is expired, redirect to login endpoint
except BearerTokenExpired:
yield rx.redirect("/login?error=token-expired)
def get_recipes_with_parmesan_cheese(self) -> typing.Generator:
try:
# From 'recipes' table select all recipes where parmesan is in list of ingredients
results = self.query().table("recipes").select("*").in_("ingredients", ["parmesan"]).execute()
# Set recipes to state
yield BaseState.setvar("recipes", results)
# If users token is expired, redirect to login endpoint
except BearerTokenExpired:
yield rx.redirect("/login?error=token-expired)
```
### Admin Queries
Chaining in .admin() after a .query() will run the query using the service role key you provided in the config file. This will bypass row level security and allows admin access during user actions within reflex that require retrieving or updating information beyond the scope of your user's permissions.
We let a user who has insert privileges upload a message. This would succeed with simple RLS policies.
```python
self.query().table("messages").insert(message_data).execute()
```
Prior to uploading you want to check all usernames prior to sending to make sure that the recipient that user has selected exists.
```python
user_exists = self.query().table("messages").select("*").eq("user_handle", recipient).limit(1).execute()
if user_exists:
continue_with_sending_message()
else:
notify_no_user_present()
```
This would fail, so instead with no change to code structure or opening up RLS policies to where we might accidentally allow a user to get access to all user's messages.
```python
user_exists = self.query().admin().table("messages").select("*").eq("user_handle", recipient).limit(1).execute()
```
This would succeed as chaining admin() to the query() swaps the user supplied JWT token with the config supplied service_role token BUT ONLY FOR THAT QUERY. This eliminates the scenario where inadvertently standard user queries may be made with a service role token.
### Query Methods
[Python API Reference | Supabase Docs](https://supabase.com/docs/reference/python/select)
- select(column: str)
- Specify column(s) to return or '*' to return all.
- insert(data: dict[str, Any] | list, return_: Literal["representation", "minimal"])
- Add new data to specified .table(). Can specify if the inserted row is returned by setting return_ to "representation".
- upsert(data: dict, return_: Literal["representation","minimal"])
- Add item to specified .table() if it doesn't exist, otherwise update item. One column must be primary key.
- update(data: Dict[str, Any], return_: Literal["representation","minimal"])
- Update rows with specified data - will match all rows by default. Use filters to update specific rows like eq(), lt(), or is().
- delete()
- Deletes rows - will match all rows by default. Use filters to specify how to select the rows to delete.
### Query Filters (Incomplete)
[Python API Reference | Supabase Docs](https://supabase.com/docs/reference/python/using-filters)
- eq(column: str, value: Any)
- Match only rows where column is equal to value.
- neq(column: str, value: Any)
- Match only rows where column is not equal to value.
- gt(column: str, value: Any)
- Match only rows where column is greater than value.
- lt(column: str, value: Any)
- Match only rows where column is less than value.
- gte(column: str, value: Any)
- Match only rows where column is greater than or equal to value.
- lte(column: str, value: Any)
- Match only rows where column is less than or equal to value.
- like(column: str, pattern: str)
- Match only rows where column matches pattern case-sensitively.
- ilike(column: str, pattern: str)
- Match only rows where column matches pattern case-insensitively.
- is_(column: str, value: Literal["null"] | bool | None)
- Match only rows where column is null or bool. Use instead of eq() for null values.
- is_not(column: str, value: Literal["null"] | bool | None)
- Match only rows where column is NOT null/bool. Use instead of neq() for null values.
- in_(column: str, values: List[Any])
- Match only rows where columsn is in the list of values.
- contains(array_column, value: List[Any] | Dict[str, Any] | str)
- Only relevant for jsonb, array, and range columns. Match only rows where column contains every element appearing in values.
- contained_by(array_column: str, value: List[Any] | Dict[str, Any] | str)
- Only relevant for jsonb, array, and range columns. Match only rows where every element appearing in column is contained by value.
### Query Modifiers (Incomplete)
[Python API Reference | Supabase Docs](https://supabase.com/docs/reference/python/using-modifiers)
- order(column: str, ascending: bool = True, nulls_first: Optional[bool] = None)
- Order the query result by column. Defaults to ascending (lowest -> highest). Use nulls_first to place nulls at top or bottom of order.
- limit(count: int)
- Limit the number of rows returned by count.
- range(start: int, end: int)
- Limit the result starting at start offset and ending at end offset.
- csv()
- Return data as a string in CSV format.
- single()
- Not implemented yet.
- maybe_single()
- Not implemented yet.
## Other Functions
- rpc(function: str, params: Dict[Any, Any])
- Calls a postgres function deployed in Supabase.
---
## Notes
Generally this module is attempting to do the dirty work of setting up a request and turning the response into a python object. I'm leaving error handling, logging, and flows up to the devs so that you can more easily integrate this into your own flows.
If there is a feature you'd like added that keeps the spirit of flexibility but adds functionality then please let me know and I'll be happy to extend this module.
Documentation and structure of this project is **very early** so expect changes as I integrate this project and test all the features thoroughly.
Raw data
{
"_id": null,
"home_page": null,
"name": "suplex",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.12",
"maintainer_email": null,
"keywords": "authentication, database, reflex, supabase",
"author": null,
"author_email": "Jeremy Medlin <jeremy.f.medlin@gmail.com>",
"download_url": "https://files.pythonhosted.org/packages/29/e2/4a378ec659f41baea1579123db411f84befeef1d1dc31ad2b3ba2f0712bd/suplex-0.2.5.tar.gz",
"platform": null,
"description": "# Suplex\n\nSimple state module to manage user auth and create database queries with the Reflex web framework.\n\n---\n\n## Install\n\nAdd Suplex to your project.\n\n```bash\n# Install uv with curl\ncurl -LsSf https://astral.sh/uv/install.sh | sh\n\n# Install uv with wget\nwget -qO- https://astral.sh/uv/install.sh | sh\n\n# Activate your venv in your base folder\ncd /path/to/project\nuv venv # or uv venv my-project-venv\nsource .venv/bin/activate # or source /my-project-venv/bin/activate\n\n# Add suplex as dependency\nuv add suplex\n```\n\n---\n\n## Configure\n\nIn your project top-level directory, where rxconfig.py is located create a .env file as follows...\n\n```bash\napi_url=\"your-api-url\"\napi_key=\"your-api-key\"\njwt_secret=\"your-jwt-secret\"\nservice_role=\"your-service-role\"\n```\n\nThese values can be retrieved from Supabase. Log In >> Choose Project >> Project Settings >> Data API\n\nThen in rxconfig.py add...\n\n```python\nfrom dotenv import load_dotenv\n\nload_dotenv()\n\napi_key = os.getenv(\"api_key\")\napi_url = os.getenv(\"api_url\")\njwt_secret = os.getenv(\"jwt_secret\")\n# service_role = os.getenv(\"service_role\") Only for admin use.\n\nconfig = rx.Config(\n # You may have a few entries here...\n suplex={\n \"api_url\": api_url,\n \"api_key\": api_key,\n \"jwt_secret\": jwt_secret\n \"let_jwt_expire\": False # (Optional: Default is False) Specify if tokens auto refresh. Can set to True for tighter/manual control of token refresh\n \"cookie_max_age\": 3600 # (Optional: Default = None) Seconds until cookie expires, otherwise is a session cookie.\n } \n) \n```\n\n---\n\n## Build into existing State Class\n\nImport Suplex, and subclass the module at the lowest layer. A BaseState class that is already built, or fresh will work. For a BaseState that is already built, make sure that you check the suplex functions and vars to ensure there aren't any collisions.\n\n```python\nfrom suplex import Suplex\n\nclass BaseState(Suplex):\n # Your class below...\n```\n\n---\n\n## Other Subclassing\n\nFor any other classes within your Reflex project, subclass your BaseState to give them access to the auth information and query methods. There shouldn't be any classes in your state requiring auth that don't inherit from the BaseState.\n\n```python\nclass OtherState(BaseState):\n # Your class below...\n```\n\n---\n\n## Auth\n\nSuplex comes with built-in vars and functions to manage users, user objects, JWT claims and more. Because front-end handling is different from project to project, you'll need to create functions for how you'd like to update your UI, redirect on success/failure, and handle exceptions.\n\n### Auth Functions\n\n- sign_up(email: str, phone: str, password: str, options: Dict[str, Any])\n \n - Sign a user up using email or phone, and password. May pass in options - email_redirect_to, data, captcha_token, and channel.\n\n- sign_in_with_password(email: str, phone: str, password: str, options: Dict[str, Any])\n \n - Sign a user in with email or phone, and password. May pass in options - captcha_token.\n \n - Returns user object dict and other info.\n\n- sign_in_with_oauth(provider: str, options: Dict[str, Any])\n \n - Returns a url to use with rx.redirect to send user to OAuth endpoint. May pass in options redirect_to, scopes, query_params, code_challenge, and code_challenge_method. Will need to manually parse the url that OAuth redirects to for storing the tokens. Use set_tokens() to set access_token and refresh_token.\n\n- exchange_code_for_session(params: Dict[str, Any])\n \n - Provide params auth_code, code_verifier for an access_token and refresh_token when using PKCE flow. Sets tokens automatically. Returns the user object and other information.\n\n- set_tokens(access_token: str, refresh_token: str)\n \n - Sets both tokens as well as setting the bearer token of the query object using the access_token. Use this if manually parsing a URL to set tokens.\n\n- reset_password_email(email: str)\n \n - Send password reset email to specified email address.\n\n- get_user()\n \n - Returns the user object of the current authenticated user from Supabase.\n\n- update_user(email: str, phone: str, password: str, user_metadata: Dict[str, Any])\n \n - Updates the current users email, phone, password, or custom data stored in user_metadata.\n \n - Returns the updated user object.\n\n- refresh_session()\n \n - Manually refreshes the authentication session using the stored refresh token.\n\n- get_settings()\n \n - Returns the authentication settings for the Supabase project.\n\n- log_out()\n \n - Clears local tokens and attempts to nullify the refresh_token to Supabase.\n\n- session_manager(event: Callable, on_failure: List[Callable] | Callable | None)\n \n - Pass an event. If users token has expired, will attempt to refresh the token. If the token is unable to be refreshed, will trigger the on_failure event. Used typically to redirect to login page on failure.\n\nCheck docstrings for params, returns and exceptions.\n\n### Exceptions\n\nAuth requests are handled using the httpx library. Suplex raises no specific exceptions, so all issues with connection will be raised via the httpx exception classes. You can specify handling specific status codes as well (see examples below).\n\n[QuickStart - HTTPX Exceptions](https://www.python-httpx.org/quickstart/#exceptions)\n\n[HTTP response status codes - HTTP | MDN](https://developer.mozilla.org/en-US/docs/Web/HTTP/Reference/Status)\n\n```python\nfrom suplex import Suplex\n\nimport typing\n\n\nclass BaseState(Suplex):\n # Login example.\n def log_in(self, email, password) -> typing.Generator:\n try:\n self.sign_in_with_password(email, password)\n except httpx.HTTPStatusError as exc:\n if exc.response.status_code == 401:\n yield rx.toast.error(\"Invalid email or password.\")\n except Exception:\n yield rx.toast.error(\"Login failed.\")\n\n # Update user example.\n def update_user_info(self, email, phone, password, user_metadata) -> typing.Generator:\n try:\n self.update_user(email, phone, password, user_metadata)\n except httpx.HTTPStatusError as exc:\n if exc.response.status_code == 403:\n # Refresh token here and try again.\n except Exception:\n yield rx.toast.error(\"Updating user info failed.\")\n\n def log_out(self):\n try:\n self.logout()\n except httpx.HTTPStatusError:\n yield rx.toast.error(\"Unable to logout remotely, local session cleared.\")\n except Exception:\n yield rx.toast.error(\"Something went wrong during logout.\")\n```\n\n### Auth Vars\n\nThere is a full set of vars that pull values from the signed JWT that gets provided from Supabase in the form of an access_token. These vars pull those claims. If you don't want to use local information and instead only want to rely on a user object pulled directly from Supabase then you will want to use the get_user() function and parse the user object directly.\n\n- user_id\n \n - str - user's id as uuid.\n\n- user_email\n \n - str - user's email.\n\n- user_phone\n \n - str = user's phone.\n\n- user_audience\n \n - str - audience that user is a part of. Currently only supports audience \"authenticated\" as the JWT checks for this audience for validation.\n\n- user_role\n \n - str - role assigned to user.\n\n- claims_issuer\n \n - str - url that issued claim.\n\n- claims_expire_at\n \n - int - unix timestamp for when token expires.\n\n- claims_issued_at\n \n - int - unix timestamp for when token was issued.\n\n- claims_session_id\n \n - str - unique session id from Supabase.\n\n- user_metadata\n \n - Dict[str, Any] - custom data within user object.\n\n- app_metadata\n \n - Dict[str, Any] - extra info stored by Supabase.\n\n- user_aal\n \n - Literal[\"aal1\" | \"aal2\"] - specifies 1 or 2 factor auth was used.\n\n- user_is_authenticated\n \n - bool - if user audience is \"authenticated\" will be True.\n\n- user_is_anonymous\n \n - bool - if user is anonymous will be True.\n\n- user_token_expired\n \n - bool - checks expiry for claims.\n\n```python\n# Frontend\ndef auth_component() -> rx.Component:\n # Show only if user is logged in.\n return rx.cond(\n BaseState.user_is_authenticated,\n shown_if_authenticated(),\n shown_if_not_authenticated()\n)\n\ndef user_info_panel() -> rx.Component:\n # Show currently logged in user info.\n return rx.flex(\n rx.text(BaseState.user_id),\n rx.text(BaseState.user_email),\n rx.text(BaseState.user_phone),\n class_name=\"flex-col items-center w-full\"\n)\n\n# Setup a page to use auth_flow. Redirects user who isn't logged in.\n@rx.page(\"/recipes\", on_load=BaseState.auth_flow)\ndef recipes() -> rx.Component:\n return rx.flex(\n rx.button(\"Get Recipes\")\n on_click=BaseState.get_recipes()\n)\n\nclass BaseState(Suplex):\n\n def auth_flow(self) -> Callable:\n if not self.user_is_authenticated:\n return rx.redirect(\"/login\")\n```\n\n---\n\n## Query\n\nQueries are built using the query() function to return an instance of the query that is then chained with other functions and then executed. Each query is it's own instance that executes in order for multiple queries to not overwrite data within Reflex's asynchronous event handler.\n\nThe style of query is similar to the official Supabase python client located at - [Python API Reference | Supabase Docs](https://supabase.com/docs/reference/python/select).\n\nIf a user's bearer token has expired, Suplex handles refreshing the session as long as within the initial rxconfig.py setup, let_jwt_expire is set to False (the default). If you prefer to manually handle jwt_expiration, or the refresh_session fails to refresh user's tokens properly, the query() function will raise a BearerTokenExpired exception.\n\n```python\nfrom suplex import Suplex, BearerTokenExpired\nimport typing\n\n\nclass BaseState(Suplex):\n ingredients: list\n recipes: list\n\n def get_all_ingredients(self) -> typing.Generator:\n try:\n ingredients = []\n # From 'recipes' table, select row ingredients\n results = self.query().table(\"recipes\").select(\"ingredients\").execute()\n for result in results:\n ingredients.extend(result[\"ingredients\"])\n # Ensure no duplicate ingredients are present\n ingredients = list(set(ingredients)\n\n # Set ingredients to State\n yield BaseState.setvar(\"ingredients\", ingredients)\n\n # If users token is expired, redirect to login endpoint\n except BearerTokenExpired:\n yield rx.redirect(\"/login?error=token-expired)\n\n\n def get_recipes_with_parmesan_cheese(self) -> typing.Generator:\n try:\n # From 'recipes' table select all recipes where parmesan is in list of ingredients\n results = self.query().table(\"recipes\").select(\"*\").in_(\"ingredients\", [\"parmesan\"]).execute()\n \n # Set recipes to state\n yield BaseState.setvar(\"recipes\", results)\n\n\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0\u00a0# If users token is expired, redirect to login endpoint\n except BearerTokenExpired:\n yield rx.redirect(\"/login?error=token-expired)\n```\n\n### Admin Queries\n\nChaining in .admin() after a .query() will run the query using the service role key you provided in the config file. This will bypass row level security and allows admin access during user actions within reflex that require retrieving or updating information beyond the scope of your user's permissions.\n\nWe let a user who has insert privileges upload a message. This would succeed with simple RLS policies.\n\n```python\nself.query().table(\"messages\").insert(message_data).execute()\n```\nPrior to uploading you want to check all usernames prior to sending to make sure that the recipient that user has selected exists.\n\n```python\nuser_exists = self.query().table(\"messages\").select(\"*\").eq(\"user_handle\", recipient).limit(1).execute()\nif user_exists:\n continue_with_sending_message()\nelse:\n notify_no_user_present()\n```\n\nThis would fail, so instead with no change to code structure or opening up RLS policies to where we might accidentally allow a user to get access to all user's messages.\n\n```python\nuser_exists = self.query().admin().table(\"messages\").select(\"*\").eq(\"user_handle\", recipient).limit(1).execute()\n```\n\nThis would succeed as chaining admin() to the query() swaps the user supplied JWT token with the config supplied service_role token BUT ONLY FOR THAT QUERY. This eliminates the scenario where inadvertently standard user queries may be made with a service role token.\n\n### Query Methods\n\n[Python API Reference | Supabase Docs](https://supabase.com/docs/reference/python/select)\n\n- select(column: str)\n \n - Specify column(s) to return or '*' to return all.\n\n- insert(data: dict[str, Any] | list, return_: Literal[\"representation\", \"minimal\"])\n \n - Add new data to specified .table(). Can specify if the inserted row is returned by setting return_ to \"representation\".\n\n- upsert(data: dict, return_: Literal[\"representation\",\"minimal\"])\n \n - Add item to specified .table() if it doesn't exist, otherwise update item. One column must be primary key.\n\n- update(data: Dict[str, Any], return_: Literal[\"representation\",\"minimal\"])\n \n - Update rows with specified data - will match all rows by default. Use filters to update specific rows like eq(), lt(), or is().\n\n- delete()\n \n - Deletes rows - will match all rows by default. Use filters to specify how to select the rows to delete.\n\n### Query Filters (Incomplete)\n\n[Python API Reference | Supabase Docs](https://supabase.com/docs/reference/python/using-filters)\n\n- eq(column: str, value: Any)\n \n - Match only rows where column is equal to value.\n\n- neq(column: str, value: Any)\n \n - Match only rows where column is not equal to value.\n\n- gt(column: str, value: Any)\n \n - Match only rows where column is greater than value.\n\n- lt(column: str, value: Any)\n \n - Match only rows where column is less than value.\n\n- gte(column: str, value: Any)\n \n - Match only rows where column is greater than or equal to value.\n\n- lte(column: str, value: Any)\n \n - Match only rows where column is less than or equal to value.\n\n- like(column: str, pattern: str)\n \n - Match only rows where column matches pattern case-sensitively.\n\n- ilike(column: str, pattern: str)\n \n - Match only rows where column matches pattern case-insensitively.\n\n- is_(column: str, value: Literal[\"null\"] | bool | None)\n \n - Match only rows where column is null or bool. Use instead of eq() for null values.\n\n- is_not(column: str, value: Literal[\"null\"] | bool | None)\n \n - Match only rows where column is NOT null/bool. Use instead of neq() for null values.\n\n- in_(column: str, values: List[Any])\n \n - Match only rows where columsn is in the list of values.\n\n- contains(array_column, value: List[Any] | Dict[str, Any] | str)\n \n - Only relevant for jsonb, array, and range columns. Match only rows where column contains every element appearing in values.\n\n- contained_by(array_column: str, value: List[Any] | Dict[str, Any] | str)\n \n - Only relevant for jsonb, array, and range columns. Match only rows where every element appearing in column is contained by value.\n\n### Query Modifiers (Incomplete)\n\n[Python API Reference | Supabase Docs](https://supabase.com/docs/reference/python/using-modifiers)\n\n- order(column: str, ascending: bool = True, nulls_first: Optional[bool] = None)\n - Order the query result by column. Defaults to ascending (lowest -> highest). Use nulls_first to place nulls at top or bottom of order.\n- limit(count: int)\n - Limit the number of rows returned by count.\n- range(start: int, end: int)\n - Limit the result starting at start offset and ending at end offset.\n- csv()\n - Return data as a string in CSV format.\n- single()\n - Not implemented yet.\n- maybe_single()\n - Not implemented yet.\n\n## Other Functions\n\n- rpc(function: str, params: Dict[Any, Any])\n \n - Calls a postgres function deployed in Supabase.\n\n---\n\n## Notes\n\nGenerally this module is attempting to do the dirty work of setting up a request and turning the response into a python object. I'm leaving error handling, logging, and flows up to the devs so that you can more easily integrate this into your own flows.\n\nIf there is a feature you'd like added that keeps the spirit of flexibility but adds functionality then please let me know and I'll be happy to extend this module.\n\nDocumentation and structure of this project is **very early** so expect changes as I integrate this project and test all the features thoroughly.\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "A Python library for interacting with Supabase REST API in Reflex applications",
"version": "0.2.5",
"project_urls": {
"Bug Tracker": "https://github.com/hjpr/suplex/issues",
"Homepage": "https://github.com/hjpr/suplex"
},
"split_keywords": [
"authentication",
" database",
" reflex",
" supabase"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "ef2d3cdfefa371f62ab774674bb853ef35af3acd78256461b29666c68f4d6ed5",
"md5": "0d85f4ea4f1d4266e05b6c1167ff05a0",
"sha256": "0dd0aa4b5b6d8b34c508f1425f180939b01b77202a7e1db9ffe5eb68bf04ea35"
},
"downloads": -1,
"filename": "suplex-0.2.5-py3-none-any.whl",
"has_sig": false,
"md5_digest": "0d85f4ea4f1d4266e05b6c1167ff05a0",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.12",
"size": 16480,
"upload_time": "2025-07-21T19:53:48",
"upload_time_iso_8601": "2025-07-21T19:53:48.382688Z",
"url": "https://files.pythonhosted.org/packages/ef/2d/3cdfefa371f62ab774674bb853ef35af3acd78256461b29666c68f4d6ed5/suplex-0.2.5-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "29e24a378ec659f41baea1579123db411f84befeef1d1dc31ad2b3ba2f0712bd",
"md5": "4131f28e59bd786dd7e5fe28b5cdfc5d",
"sha256": "e34cf0c62699b1da2d2cf206ae850fbb3753dae5f962e573417fdaa0468fb731"
},
"downloads": -1,
"filename": "suplex-0.2.5.tar.gz",
"has_sig": false,
"md5_digest": "4131f28e59bd786dd7e5fe28b5cdfc5d",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.12",
"size": 15859,
"upload_time": "2025-07-21T19:53:49",
"upload_time_iso_8601": "2025-07-21T19:53:49.639791Z",
"url": "https://files.pythonhosted.org/packages/29/e2/4a378ec659f41baea1579123db411f84befeef1d1dc31ad2b3ba2f0712bd/suplex-0.2.5.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-07-21 19:53:49",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "hjpr",
"github_project": "suplex",
"travis_ci": false,
"coveralls": false,
"github_actions": false,
"lcname": "suplex"
}
