laipeng/reflex
1
0
Fork 0
Code Issues Pull Requests Actions Packages Projects Releases Wiki Activity

Refactor utils module (#666)

Browse Source
This commit is contained in:
iron3oxide 2023-03-16 22:59:27 +01:00 committed by GitHub
parent 300f4ca9cb
commit 7067baf176
No known key found for this signature in database
GPG Key ID: 4AEE18F83AFDEB23
43 changed files with 2036 additions and 1848 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

33
pynecone/app.py
View File

@ -6,16 +6,24 @@ from fastapi import FastAPI, UploadFile
from fastapi.middleware import cors
from socketio import ASGIApp, AsyncNamespace, AsyncServer
from pynecone import constants, utils
from pynecone import constants
from pynecone.base import Base
from pynecone.compiler import compiler
from pynecone.compiler import utils as compiler_utils
from pynecone.components.component import Component, ComponentStyle
from pynecone.config import get_config
from pynecone.event import Event, EventHandler
from pynecone.middleware import HydrateMiddleware, Middleware
from pynecone.model import Model
from pynecone.route import DECORATED_ROUTES
from pynecone.route import (
DECORATED_ROUTES,
catchall_in_route,
catchall_prefix,
get_route_args,
verify_route_validity,
)
from pynecone.state import DefaultState, Delta, State, StateManager, StateUpdate
from pynecone.utils import format
# Define custom types.
ComponentCallable = Callable[[], Component]
@ -65,7 +73,7 @@ class App(Base):
super().__init__(*args, **kwargs)
# Get the config
config = utils.get_config()
config = get_config()
# Add middleware.
self.middleware.append(HydrateMiddleware())
@ -233,10 +241,10 @@ class App(Base):
route = component.__name__
# Check if the route given is valid
utils.verify_route_validity(route)
verify_route_validity(route)
# Apply dynamic args to the route.
self.state.setup_dynamic_args(utils.get_route_args(route))
self.state.setup_dynamic_args(get_route_args(route))
# Generate the component if it is a callable.
try:
@ -261,7 +269,7 @@ class App(Base):
component.children.extend(script_tags)
# Format the route.
route = utils.format_route(route)
route = format.format_route(route)
# Add the page.
self._check_routes_conflict(route)
@ -281,7 +289,7 @@ class App(Base):
Args:
new_route: the route being newly added.
"""
newroute_catchall = utils.catchall_in_route(new_route)
newroute_catchall = catchall_in_route(new_route)
if not newroute_catchall:
return
@ -293,11 +301,11 @@ class App(Base):
f"You cannot define a route with the same specificity as a optional catch-all route ('{route}' and '{new_route}')"
)
route_catchall = utils.catchall_in_route(route)
route_catchall = catchall_in_route(route)
if (
route_catchall
and newroute_catchall
and utils.catchall_prefix(route) == utils.catchall_prefix(new_route)
and catchall_prefix(route) == catchall_prefix(new_route)
):
raise ValueError(
f"You cannot use multiple catchall for the same dynamic route ({route} !== {new_route})"
@ -333,7 +341,7 @@ class App(Base):
component, title=title, image=image, description=description, meta=meta
)
froute = utils.format_route
froute = format.format_route
if (froute(constants.ROOT_404) not in self.pages) and (
not any(page.startswith("[[...") for page in self.pages)
):
@ -356,7 +364,7 @@ class App(Base):
self.add_page(render, **kwargs)
# Get the env mode.
config = utils.get_config()
config = get_config()
if config.env != constants.Env.DEV and not force_compile:
print("Skipping compilation in non-dev mode.")
return
@ -405,7 +413,7 @@ async def process(
# Get the state for the session.
state = app.state_manager.get_state(event.token)
formatted_params = utils.format_query_params(event.router_data)
formatted_params = format.format_query_params(event.router_data)
# Pass router_data to the state of the App.
state.router_data = event.router_data
@ -523,6 +531,7 @@ class EventNamespace(AsyncNamespace):
# Get the event environment.
assert self.app.sio is not None
environ = self.app.sio.get_environ(sid, self.namespace)
assert environ is not None
# Get the client headers.
headers = {

9
pynecone/compiler/compiler.py
View File

@ -7,12 +7,13 @@ from typing import Callable, List, Set, Tuple, Type
from pynecone import constants
from pynecone.compiler import templates, utils
from pynecone.components.component import Component, CustomComponent, ImportDict
from pynecone.components.component import Component, CustomComponent
from pynecone.state import State
from pynecone.style import Style
from pynecone.utils import imports, path_ops
# Imports to be included in every Pynecone app.
DEFAULT_IMPORTS: ImportDict = {
DEFAULT_IMPORTS: imports.ImportDict = {
"react": {"useEffect", "useRef", "useState"},
"next/router": {"useRouter"},
f"/{constants.STATE_PATH}": {"connect", "updateState", "uploadFiles", "E"},
@ -64,7 +65,7 @@ def _compile_page(component: Component, state: Type[State]) -> str:
# Compile the code to render the component.
return templates.PAGE(
imports=utils.compile_imports(imports),
custom_code=templates.join(component.get_custom_code()),
custom_code=path_ops.join(component.get_custom_code()),
constants=utils.compile_constants(),
state=utils.compile_state(state),
events=utils.compile_events(state),
@ -97,7 +98,7 @@ def _compile_components(components: Set[CustomComponent]) -> str:
# Compile the components page.
return templates.COMPONENTS(
imports=utils.compile_imports(imports),
components=templates.join(component_defs),
components=path_ops.join(component_defs),
)

18
pynecone/compiler/templates.py
View File

@ -3,7 +3,7 @@
from typing import Optional, Set
from pynecone import constants
from pynecone.utils import join
from pynecone.utils import path_ops
# Template for the Pynecone config file.
PCCONFIG = f"""import pynecone as pc
@ -37,7 +37,7 @@ def format_import(lib: str, default: str = "", rest: Optional[Set[str]] = None)
if not lib:
assert not default, "No default field allowed for empty library."
assert rest is not None and len(rest) > 0, "No fields to import."
return join([IMPORT_LIB(lib=lib) for lib in sorted(rest)])
return path_ops.join([IMPORT_LIB(lib=lib) for lib in sorted(rest)])
# Handle importing from a library.
rest = rest or set()
@ -52,7 +52,7 @@ def format_import(lib: str, default: str = "", rest: Optional[Set[str]] = None)
# Code to render a NextJS Document root.
DOCUMENT_ROOT = join(
DOCUMENT_ROOT = path_ops.join(
[
"{imports}",
"export default function Document() {{",
@ -67,7 +67,7 @@ DOCUMENT_ROOT = join(
THEME = "export default {theme}".format
# Code to render a single NextJS page.
PAGE = join(
PAGE = path_ops.join(
[
"{imports}",
"{custom_code}",
@ -84,7 +84,7 @@ PAGE = join(
).format
# Code to render a single exported custom component.
COMPONENT = join(
COMPONENT = path_ops.join(
[
"export const {name} = memo(({{{props}}}) => (",
"{render}",
@ -93,7 +93,7 @@ COMPONENT = join(
).format
# Code to render the custom components page.
COMPONENTS = join(
COMPONENTS = path_ops.join(
[
"{imports}",
"{components}",
@ -138,7 +138,7 @@ def format_state(
# Events.
EVENT_ENDPOINT = constants.Endpoint.EVENT.name
EVENT_FN = join(
EVENT_FN = path_ops.join(
[
"const Event = events => {set_state}({{",
" ...{state},",
@ -146,7 +146,7 @@ EVENT_FN = join(
"}})",
]
).format
UPLOAD_FN = join(
UPLOAD_FN = path_ops.join(
[
"const File = files => {set_state}({{",
" ...{state},",
@ -165,7 +165,7 @@ STATE = constants.STATE
EVENTS = constants.EVENTS
SET_RESULT = format_state_setter(RESULT)
READY = f"const {{ isReady }} = {ROUTER};"
USE_EFFECT = join(
USE_EFFECT = path_ops.join(
[
"useEffect(() => {{",
" if(!isReady) {{",

30
pynecone/compiler/utils.py
View File

@ -4,7 +4,7 @@ import json
import os
from typing import Dict, List, Optional, Set, Tuple, Type
from pynecone import constants, utils
from pynecone import constants
from pynecone.compiler import templates
from pynecone.components.base import (
Body,
@ -20,12 +20,14 @@ from pynecone.components.base import (
Script,
Title,
)
from pynecone.components.component import Component, CustomComponent, ImportDict
from pynecone.components.component import Component, CustomComponent
from pynecone.event import get_hydrate_event
from pynecone.state import State
from pynecone.style import Style
from pynecone.utils import format, imports, path_ops
# To re-export this function.
merge_imports = utils.merge_imports
merge_imports = imports.merge_imports
def compile_import_statement(lib: str, fields: Set[str]) -> str:
@ -52,7 +54,7 @@ def compile_import_statement(lib: str, fields: Set[str]) -> str:
return templates.format_import(lib=lib, default=default, rest=rest)
def compile_imports(imports: ImportDict) -> str:
def compile_imports(imports: imports.ImportDict) -> str:
"""Compile an import dict.
Args:
@ -61,7 +63,7 @@ def compile_imports(imports: ImportDict) -> str:
Returns:
The compiled import dict.
"""
return templates.join(
return path_ops.join(
[compile_import_statement(lib, fields) for lib, fields in imports.items()]
)
@ -85,7 +87,7 @@ def compile_constants() -> str:
Returns:
A string of all the compiled constants.
"""
return templates.join(
return path_ops.join(
[
compile_constant_declaration(name=endpoint.name, value=endpoint.get_url())
for endpoint in constants.Endpoint
@ -105,11 +107,11 @@ def compile_state(state: Type[State]) -> str:
initial_state = state().dict()
initial_state.update(
{
"events": [{"name": utils.get_hydrate_event(state)}],
"events": [{"name": get_hydrate_event(state)}],
"files": [],
}
)
initial_state = utils.format_state(initial_state)
initial_state = format.format_state(initial_state)
synced_state = templates.format_state(
state=state.get_name(), initial_state=json.dumps(initial_state)
)
@ -126,7 +128,7 @@ def compile_state(state: Type[State]) -> str:
socket = templates.SOCKET
ready = templates.READY
color_toggle = templates.COLORTOGGLE
return templates.join([synced_state, result, router, socket, ready, color_toggle])
return path_ops.join([synced_state, result, router, socket, ready, color_toggle])
def compile_events(state: Type[State]) -> str:
@ -140,7 +142,7 @@ def compile_events(state: Type[State]) -> str:
"""
state_name = state.get_name()
state_setter = templates.format_state_setter(state_name)
return templates.join(
return path_ops.join(
[
templates.EVENT_FN(state=state_name, set_state=state_setter),
templates.UPLOAD_FN(state=state_name, set_state=state_setter),
@ -177,7 +179,9 @@ def compile_render(component: Component) -> str:
return component.render()
def compile_custom_component(component: CustomComponent) -> Tuple[str, ImportDict]:
def compile_custom_component(
component: CustomComponent,
) -> Tuple[str, imports.ImportDict]:
"""Compile a custom component.
Args:
@ -322,7 +326,7 @@ def write_page(path: str, code: str):
path: The path to write the code to.
code: The code to write.
"""
utils.mkdir(os.path.dirname(path))
path_ops.mkdir(os.path.dirname(path))
with open(path, "w", encoding="utf-8") as f:
f.write(code)
@ -343,4 +347,4 @@ def empty_dir(path: str, keep_files: Optional[List[str]] = None):
directory_contents = os.listdir(path)
for element in directory_contents:
if element not in keep_files:
utils.rm(os.path.join(path, element))
path_ops.rm(os.path.join(path, element))

48
pynecone/components/component.py
View File

@ -7,7 +7,7 @@ from abc import ABC
from functools import wraps
from typing import Any, Callable, Dict, List, Optional, Set, Type, Union
from pynecone import constants, utils
from pynecone import constants
from pynecone.base import Base
from pynecone.components.tags import Tag
from pynecone.event import (
@ -16,12 +16,14 @@ from pynecone.event import (
EventChain,
EventHandler,
EventSpec,
call_event_fn,
call_event_handler,
get_handler_args,
)
from pynecone.style import Style
from pynecone.utils import format, imports, path_ops, types
from pynecone.var import BaseVar, Var
ImportDict = Dict[str, Set[str]]
class Component(Base, ABC):
"""The base class for all Pynecone components."""
@ -72,7 +74,7 @@ class Component(Base, ABC):
continue
# Set default values for any props.
if utils._issubclass(field.type_, Var):
if types._issubclass(field.type_, Var):
field.required = False
field.default = Var.create(field.default)
@ -109,7 +111,7 @@ class Component(Base, ABC):
continue
# Check whether the key is a component prop.
if utils._issubclass(field_type, Var):
if types._issubclass(field_type, Var):
try:
# Try to create a var from the value.
kwargs[key] = Var.create(value)
@ -125,7 +127,7 @@ class Component(Base, ABC):
# If it is not a valid var, check the base types.
passed_type = type(value)
expected_type = fields[key].outer_type_
if not utils._issubclass(passed_type, expected_type):
if not types._issubclass(passed_type, expected_type):
raise TypeError(
f"Invalid var passed for prop {key}, expected type {expected_type}, got value {value} of type {passed_type}."
)
@ -200,7 +202,7 @@ class Component(Base, ABC):
for v in value:
if isinstance(v, EventHandler):
# Call the event handler to get the event.
event = utils.call_event_handler(v, arg)
event = call_event_handler(v, arg)
# Check that the event handler takes no args if it's uncontrolled.
if not is_controlled_event and len(event.args) > 0:
@ -215,13 +217,13 @@ class Component(Base, ABC):
events.append(v)
elif isinstance(v, Callable):
# Call the lambda to get the event chain.
events.extend(utils.call_event_fn(v, arg))
events.extend(call_event_fn(v, arg))
else:
raise ValueError(f"Invalid event: {v}")
# If the input is a callable, create an event chain.
elif isinstance(value, Callable):
events = utils.call_event_fn(value, arg)
events = call_event_fn(value, arg)
# Otherwise, raise an error.
else:
@ -233,7 +235,7 @@ class Component(Base, ABC):
EventSpec(
handler=e.handler,
local_args=(EVENT_ARG.name,),
args=utils.get_handler_args(e, arg),
args=get_handler_args(e, arg),
)
for e in events
]
@ -333,7 +335,7 @@ class Component(Base, ABC):
# Validate all the children.
for child in children:
# Make sure the child is a valid type.
if not utils._isinstance(child, ComponentChild):
if not types._isinstance(child, ComponentChild):
raise TypeError(
"Children of Pynecone components must be other components, "
"state vars, or primitive Python types. "
@ -392,7 +394,7 @@ class Component(Base, ABC):
id=self.id,
class_name=self.class_name,
).set(
contents=utils.join(
contents=path_ops.join(
[str(tag.contents)] + [child.render() for child in self.children]
).strip(),
)
@ -427,20 +429,20 @@ class Component(Base, ABC):
# Return the code.
return code
def _get_imports(self) -> ImportDict:
def _get_imports(self) -> imports.ImportDict:
if self.library is not None and self.tag is not None:
alias = self.get_alias()
tag = self.tag if alias is None else " as ".join([self.tag, alias])
return {self.library: {tag}}
return {}
def get_imports(self) -> ImportDict:
def get_imports(self) -> imports.ImportDict:
"""Get all the libraries and fields that are used by the component.
Returns:
The import dict with the required imports.
"""
return utils.merge_imports(
return imports.merge_imports(
self._get_imports(), *[child.get_imports() for child in self.children]
)
@ -467,7 +469,7 @@ class Component(Base, ABC):
# Map from component to styling.
ComponentStyle = Dict[Union[str, Type[Component]], Any]
ComponentChild = Union[utils.PrimitiveType, Var, Component]
ComponentChild = Union[types.PrimitiveType, Var, Component]
class CustomComponent(Component):
@ -495,7 +497,7 @@ class CustomComponent(Component):
self.style = Style()
# Set the tag to the name of the function.
self.tag = utils.to_title_case(self.component_fn.__name__)
self.tag = format.to_title_case(self.component_fn.__name__)
# Set the props.
props = typing.get_type_hints(self.component_fn)
@ -503,19 +505,19 @@ class CustomComponent(Component):
if key not in props:
continue
type_ = props[key]
if utils._issubclass(type_, EventChain):
if types._issubclass(type_, EventChain):
value = self._create_event_chain(key, value)
self.props[utils.to_camel_case(key)] = value
self.props[format.to_camel_case(key)] = value
continue
type_ = utils.get_args(type_)[0]
if utils._issubclass(type_, Base):
type_ = types.get_args(type_)[0]
if types._issubclass(type_, Base):
try:
value = BaseVar(name=value.json(), type_=type_, is_local=True)
except Exception:
value = Var.create(value)
else:
value = Var.create(value, is_string=type(value) is str)
self.props[utils.to_camel_case(key)] = value
self.props[format.to_camel_case(key)] = value
def __eq__(self, other: Any) -> bool:
"""Check if the component is equal to another.
@ -586,7 +588,7 @@ class CustomComponent(Component):
return [
BaseVar(
name=name,
type_=prop.type_ if utils._isinstance(prop, Var) else type(prop),
type_=prop.type_ if types._isinstance(prop, Var) else type(prop),
)
for name, prop in self.props.items()
]

14
pynecone/components/datadisplay/code.py
View File

@ -2,10 +2,10 @@
from typing import Dict
from pynecone import utils
from pynecone.components.component import Component, ImportDict
from pynecone.components.component import Component
from pynecone.components.libs.chakra import ChakraComponent
from pynecone.style import Style
from pynecone.utils import imports
from pynecone.var import Var
# Path to the prism styles.
@ -40,13 +40,13 @@ class CodeBlock(Component):
# Props passed down to the code tag.
code_tag_props: Var[Dict[str, str]]
def _get_imports(self) -> ImportDict:
imports = super()._get_imports()
def _get_imports(self) -> imports.ImportDict:
merged_imports = super()._get_imports()
if self.theme is not None:
imports = utils.merge_imports(
imports, {PRISM_STYLES_PATH: {self.theme.name}}
merged_imports = imports.merge_imports(
merged_imports, {PRISM_STYLES_PATH: {self.theme.name}}
)
return imports
return merged_imports
@classmethod
def create(cls, *children, **props):

20
pynecone/components/datadisplay/datatable.py
View File

@ -2,9 +2,9 @@
from typing import Any, List, Optional
from pynecone import utils
from pynecone.components.component import Component, ImportDict
from pynecone.components.component import Component
from pynecone.components.tags import Tag
from pynecone.utils import format, imports, types
from pynecone.var import BaseVar, Var
@ -64,8 +64,8 @@ class DataTable(Gridjs):
# If data is a pandas dataframe and columns are provided throw an error.
if (
utils.is_dataframe(type(data))
or (isinstance(data, Var) and utils.is_dataframe(data.type_))
types.is_dataframe(type(data))
or (isinstance(data, Var) and types.is_dataframe(data.type_))
) and props.get("columns"):
raise ValueError(
"Cannot pass in both a pandas dataframe and columns to the data_table component."
@ -86,8 +86,8 @@ class DataTable(Gridjs):
**props,
)
def _get_imports(self) -> ImportDict:
return utils.merge_imports(
def _get_imports(self) -> imports.ImportDict:
return imports.merge_imports(
super()._get_imports(), {"": {"gridjs/dist/theme/mermaid.css"}}
)
@ -96,23 +96,23 @@ class DataTable(Gridjs):
if isinstance(self.data, Var):
self.columns = BaseVar(
name=f"{self.data.name}.columns"
if utils.is_dataframe(self.data.type_)
if types.is_dataframe(self.data.type_)
else f"{self.columns.name}",
type_=List[Any],
state=self.data.state,
)
self.data = BaseVar(
name=f"{self.data.name}.data"
if utils.is_dataframe(self.data.type_)
if types.is_dataframe(self.data.type_)
else f"{self.data.name}",
type_=List[List[Any]],
state=self.data.state,
)
# If given a pandas df break up the data and columns
if utils.is_dataframe(type(self.data)):
if types.is_dataframe(type(self.data)):
self.columns = Var.create(list(self.data.columns.values.tolist())) # type: ignore
self.data = Var.create(utils.format_dataframe_values(self.data)) # type: ignore
self.data = Var.create(format.format_dataframe_values(self.data)) # type: ignore
# Render the table.
return super()._render()

4
pynecone/components/forms/radio.py
View File

@ -3,12 +3,12 @@
from typing import Any, Dict, List
from pynecone import utils
from pynecone.components.component import Component
from pynecone.components.layout.foreach import Foreach
from pynecone.components.libs.chakra import ChakraComponent
from pynecone.components.typography.text import Text
from pynecone.event import EVENT_ARG
from pynecone.utils import types
from pynecone.var import Var
@ -47,7 +47,7 @@ class RadioGroup(ChakraComponent):
if (
len(children) == 1
and isinstance(children[0], Var)
and utils._issubclass(children[0].type_, List)
and types._issubclass(children[0].type_, List)
):
children = [Foreach.create(children[0], lambda item: Radio.create(item))]
return super().create(*children, **props)

4
pynecone/components/forms/select.py
View File

@ -2,11 +2,11 @@
from typing import Any, Dict, List
from pynecone import utils
from pynecone.components.component import EVENT_ARG, Component
from pynecone.components.layout.foreach import Foreach
from pynecone.components.libs.chakra import ChakraComponent
from pynecone.components.typography.text import Text
from pynecone.utils import types
from pynecone.var import Var
@ -75,7 +75,7 @@ class Select(ChakraComponent):
if (
len(children) == 1
and isinstance(children[0], Var)
and utils._issubclass(children[0].type_, List)
and types._issubclass(children[0].type_, List)
):
children = [Foreach.create(children[0], lambda item: Option.create(item))]
return super().create(*children, **props)

4
pynecone/components/layout/cond.py
View File

@ -3,10 +3,10 @@ from __future__ import annotations
from typing import Any, Optional
from pynecone import utils
from pynecone.components.component import Component
from pynecone.components.layout.fragment import Fragment
from pynecone.components.tags import CondTag, Tag
from pynecone.utils import format
from pynecone.var import Var
@ -93,7 +93,7 @@ def cond(condition: Any, c1: Any, c2: Any = None):
# Create the conditional var.
return BaseVar(
name=utils.format_cond(
name=format.format_cond(
cond=cond_var.full_name,
true_value=c1,
false_value=c2,

4
pynecone/components/media/icon.py
View File

@ -1,7 +1,7 @@
"""An icon component."""
from pynecone import utils
from pynecone.components.component import Component
from pynecone.utils import format
class ChakraIconComponent(Component):
@ -42,7 +42,7 @@ class Icon(ChakraIconComponent):
raise ValueError(
f"Invalid icon tag: {props['tag']}. Please use one of the following: {ICON_LIST}"
)
props["tag"] = utils.to_title_case(props["tag"]) + "Icon"
props["tag"] = format.to_title_case(props["tag"]) + "Icon"
return super().create(*children, **props)

4
pynecone/components/tags/cond_tag.py
View File

@ -2,8 +2,8 @@
from typing import Any
from pynecone import utils
from pynecone.components.tags.tag import Tag
from pynecone.utils import format
from pynecone.var import Var
@ -26,7 +26,7 @@ class CondTag(Tag):
The React code to render the tag.
"""
assert self.cond is not None, "The condition must be set."
return utils.format_cond(
return format.format_cond(
cond=self.cond.full_name,
true_value=self.true_value,
false_value=self.false_value,

8
pynecone/components/tags/iter_tag.py
View File

@ -4,9 +4,9 @@ from __future__ import annotations
import inspect
from typing import TYPE_CHECKING, Any, Callable, List
from pynecone import utils
from pynecone.components.tags.tag import Tag
from pynecone.var import BaseVar, Var
from pynecone.utils import format
from pynecone.var import BaseVar, Var, get_unique_variable_name
if TYPE_CHECKING:
from pynecone.components.component import Component
@ -95,12 +95,12 @@ class IterTag(Tag):
except Exception:
type_ = Any
arg = BaseVar(
name=utils.get_unique_variable_name(),
name=get_unique_variable_name(),
type_=type_,
)
index_arg = self.get_index_var_arg()
component = self.render_component(self.render_fn, arg)
return utils.wrap(
return format.wrap(
f"{self.iterable.full_name}.map(({arg.name}, {index_arg}) => {component})",
"{",
)

28
pynecone/components/tags/tag.py
View File

@ -10,9 +10,9 @@ from typing import TYPE_CHECKING, Any, Dict, Optional, Set, Tuple, Union
from plotly.graph_objects import Figure
from plotly.io import to_json
from pynecone import utils
from pynecone.base import Base
from pynecone.event import EventChain
from pynecone.utils import format, types
from pynecone.var import Var
if TYPE_CHECKING:
@ -68,7 +68,7 @@ class Tag(Base):
if not prop.is_local or prop.is_string:
return str(prop)
if issubclass(prop.type_, str):
return utils.json_dumps(prop.full_name)
return format.json_dumps(prop.full_name)
prop = prop.full_name
# Handle event props.
@ -77,18 +77,18 @@ class Tag(Base):
if len(prop.events) == 1 and prop.events[0].upload:
# Special case for upload events.
event = utils.format_upload_event(prop.events[0])
event = format.format_upload_event(prop.events[0])
else:
# All other events.
chain = ",".join([utils.format_event(event) for event in prop.events])
chain = ",".join([format.format_event(event) for event in prop.events])
event = f"Event([{chain}])"
prop = f"({local_args}) => {event}"
# Handle other types.
elif isinstance(prop, str):
if utils.is_wrapped(prop, "{"):
if format.is_wrapped(prop, "{"):
return prop
return utils.json_dumps(prop)
return format.json_dumps(prop)
elif isinstance(prop, Figure):
prop = json.loads(to_json(prop))["data"] # type: ignore
@ -103,7 +103,7 @@ class Tag(Base):
}
# Dump the prop as JSON.
prop = utils.json_dumps(prop)
prop = format.json_dumps(prop)
# This substitution is necessary to unwrap var values.
prop = re.sub('"{', "", prop)
@ -112,7 +112,7 @@ class Tag(Base):
# Wrap the variable in braces.
assert isinstance(prop, str), "The prop must be a string."
return utils.wrap(prop, "{", check_first=False)
return format.wrap(prop, "{", check_first=False)
def format_props(self) -> str:
"""Format the tag's props.
@ -149,7 +149,7 @@ class Tag(Base):
if len(self.contents) == 0:
# If there is no inner content, we don't need a closing tag.
tag_str = utils.wrap(f"{self.name}{props_str}/", "<")
tag_str = format.wrap(f"{self.name}{props_str}/", "<")
else:
if self.args is not None:
# If there are args, wrap the tag in a function call.
@ -158,9 +158,9 @@ class Tag(Base):
else:
contents = self.contents
# Otherwise wrap it in opening and closing tags.
open = utils.wrap(f"{self.name}{props_str}", "<")
close = utils.wrap(f"/{self.name}", "<")
tag_str = utils.wrap(contents, open, close)
open = format.wrap(f"{self.name}{props_str}", "<")
close = format.wrap(f"/{self.name}", "<")
tag_str = format.wrap(contents, open, close)
return tag_str
@ -175,8 +175,8 @@ class Tag(Base):
"""
self.props.update(
{
utils.to_camel_case(name): prop
if utils._isinstance(prop, Union[EventChain, dict])
format.to_camel_case(name): prop
if types._isinstance(prop, Union[EventChain, dict])
else Var.create(prop)
for name, prop in kwargs.items()
if self.is_valid_prop(prop)

4
pynecone/components/tags/tagless.py
View File

@ -1,7 +1,7 @@
"""A tag with no tag."""
from pynecone import utils
from pynecone.components.tags import Tag
from pynecone.utils import format
class Tagless(Tag):
@ -14,7 +14,7 @@ class Tagless(Tag):
The string representation of the tag.
"""
out = self.contents
space = utils.wrap(" ", "{")
space = format.wrap(" ", "{")
if len(self.contents) > 0 and self.contents[0] == " ":
out = space + out
if len(self.contents) > 0 and self.contents[-1] == " ":

4
pynecone/components/typography/markdown.py
View File

@ -3,8 +3,8 @@
import textwrap
from typing import List, Union
from pynecone import utils
from pynecone.components.component import Component
from pynecone.utils import types
from pynecone.var import BaseVar, Var
@ -26,7 +26,7 @@ class Markdown(Component):
Returns:
The markdown component.
"""
assert len(children) == 1 and utils._isinstance(
assert len(children) == 1 and types._isinstance(
children[0], Union[str, Var]
), "Markdown component must have exactly one child containing the markdown source."

17
pynecone/config.py
View File

@ -1,5 +1,7 @@
"""The Pynecone config."""
import os
import sys
from typing import List, Optional
from pynecone import constants
@ -61,3 +63,18 @@ class Config(Base):
# The maximum size of a message when using the polling backend transport.
polling_max_http_buffer_size: Optional[int] = constants.POLLING_MAX_HTTP_BUFFER_SIZE
def get_config() -> Config:
"""Get the app config.
Returns:
The app config.
"""
from pynecone.config import Config
sys.path.append(os.getcwd())
try:
return __import__(constants.CONFIG_MODULE).config
except ImportError:
return Config(app_name="") # type: ignore

10
pynecone/constants.py
View File

@ -188,10 +188,10 @@ class Endpoint(Enum):
The full URL for the endpoint.
"""
# Import here to avoid circular imports.
from pynecone import utils
from pynecone.config import get_config
# Get the API URL from the config.
config = utils.get_config()
config = get_config()
url = "".join([config.api_url, str(self)])
# The event endpoint is a websocket.
@ -241,10 +241,10 @@ class Transports(Enum):
The transports config for the backend.
"""
# Import here to avoid circular imports.
from pynecone import utils
from pynecone.config import get_config
# Get the transports from the config.
config = utils.get_config()
# Get the API URL from the config.
config = get_config()
return str(config.backend_transports)

176
pynecone/event.py
View File

@ -2,10 +2,11 @@
from __future__ import annotations
import inspect
from typing import Any, Callable, Dict, List, Set, Tuple
from typing import Any, Callable, Dict, List, Optional, Set, Tuple, Union
from pynecone import utils
from pynecone import constants
from pynecone.base import Base
from pynecone.utils import format
from pynecone.var import BaseVar, Var
@ -67,7 +68,7 @@ class EventHandler(Base):
# Otherwise, convert to JSON.
try:
values.append(utils.json_dumps(arg))
values.append(format.json_dumps(arg))
except TypeError as e:
raise TypeError(
f"Arguments to event handlers must be Vars or JSON-serializable. Got {arg} of type {type(arg)}."
@ -195,6 +196,175 @@ def window_alert(message: str) -> EventSpec:
)
def get_event(state, event):
"""Get the event from the given state.
Args:
state: The state.
event: The event.
Returns:
The event.
"""
return f"{state.get_name()}.{event}"
def get_hydrate_event(state) -> str:
"""Get the name of the hydrate event for the state.
Args:
state: The state.
Returns:
The name of the hydrate event.
"""
return get_event(state, constants.HYDRATE)
def call_event_handler(event_handler: EventHandler, arg: Var) -> EventSpec:
"""Call an event handler to get the event spec.
This function will inspect the function signature of the event handler.
If it takes in an arg, the arg will be passed to the event handler.
Otherwise, the event handler will be called with no args.
Args:
event_handler: The event handler.
arg: The argument to pass to the event handler.
Returns:
The event spec from calling the event handler.
"""
args = inspect.getfullargspec(event_handler.fn).args
if len(args) == 1:
return event_handler()
assert (
len(args) == 2
), f"Event handler {event_handler.fn} must have 1 or 2 arguments."
return event_handler(arg)
def call_event_fn(fn: Callable, arg: Var) -> List[EventSpec]:
"""Call a function to a list of event specs.
The function should return either a single EventSpec or a list of EventSpecs.
If the function takes in an arg, the arg will be passed to the function.
Otherwise, the function will be called with no args.
Args:
fn: The function to call.
arg: The argument to pass to the function.
Returns:
The event specs from calling the function.
Raises:
ValueError: If the lambda has an invalid signature.
"""
# Import here to avoid circular imports.
from pynecone.event import EventHandler, EventSpec
# Get the args of the lambda.
args = inspect.getfullargspec(fn).args
# Call the lambda.
if len(args) == 0:
out = fn()
elif len(args) == 1:
out = fn(arg)
else:
raise ValueError(f"Lambda {fn} must have 0 or 1 arguments.")
# Convert the output to a list.
if not isinstance(out, List):
out = [out]
# Convert any event specs to event specs.
events = []
for e in out:
# Convert handlers to event specs.
if isinstance(e, EventHandler):
if len(args) == 0:
e = e()
elif len(args) == 1:
e = e(arg)
# Make sure the event spec is valid.
if not isinstance(e, EventSpec):
raise ValueError(f"Lambda {fn} returned an invalid event spec: {e}.")
# Add the event spec to the chain.
events.append(e)
# Return the events.
return events
def get_handler_args(event_spec: EventSpec, arg: Var) -> Tuple[Tuple[str, str], ...]:
"""Get the handler args for the given event spec.
Args:
event_spec: The event spec.
arg: The controlled event argument.
Returns:
The handler args.
Raises:
ValueError: If the event handler has an invalid signature.
"""
args = inspect.getfullargspec(event_spec.handler.fn).args
if len(args) < 2:
raise ValueError(
f"Event handler has an invalid signature, needed a method with a parameter, got {event_spec.handler}."
)
return event_spec.args if len(args) > 2 else ((args[1], arg.name),)
def fix_events(
events: Optional[List[Union[EventHandler, EventSpec]]], token: str
) -> List[Event]:
"""Fix a list of events returned by an event handler.
Args:
events: The events to fix.
token: The user token.
Returns:
The fixed events.
"""
from pynecone.event import Event, EventHandler, EventSpec
# If the event handler returns nothing, return an empty list.
if events is None:
return []
# If the handler returns a single event, wrap it in a list.
if not isinstance(events, List):
events = [events]
# Fix the events created by the handler.
out = []
for e in events:
# Otherwise, create an event from the event spec.
if isinstance(e, EventHandler):
e = e()
assert isinstance(e, EventSpec), f"Unexpected event type, {type(e)}."
name = format.format_event_handler(e.handler)
payload = dict(e.args)
# Create an event and append it to the list.
out.append(
Event(
token=token,
name=name,
payload=payload,
)
)
return out
# A set of common event triggers.
EVENT_TRIGGERS: Set[str] = {
"on_focus",

11
pynecone/middleware/hydrate_middleware.py
View File

@ -3,10 +3,11 @@ from __future__ import annotations
from typing import TYPE_CHECKING, Optional
from pynecone import constants, utils
from pynecone.event import Event, EventHandler
from pynecone import constants
from pynecone.event import Event, EventHandler, get_hydrate_event
from pynecone.middleware.middleware import Middleware
from pynecone.state import Delta, State
from pynecone.utils import format
if TYPE_CHECKING:
from pynecone.app import App
@ -26,7 +27,7 @@ class HydrateMiddleware(Middleware):
Returns:
An optional state to return.
"""
if event.name == utils.get_hydrate_event(state):
if event.name == get_hydrate_event(state):
route = event.router_data.get(constants.RouteVar.PATH, "")
if route == "/":
load_event = app.load_events.get(constants.INDEX_ROUTE)
@ -41,7 +42,7 @@ class HydrateMiddleware(Middleware):
self.execute_load_event(state, single_event)
else:
self.execute_load_event(state, load_event)
return utils.format_state({state.get_name(): state.dict()})
return format.format_state({state.get_name(): state.dict()})
def execute_load_event(self, state: State, load_event: EventHandler) -> None:
"""Execute single load event.
@ -50,6 +51,6 @@ class HydrateMiddleware(Middleware):
state: The client state.
load_event: A single load event to execute.
"""
substate_path = utils.format_event_handler(load_event).split(".")
substate_path = format.format_event_handler(load_event).split(".")
ex_state = state.get_substate(substate_path[:-1])
load_event.fn(ex_state)

4
pynecone/model.py
View File

@ -2,8 +2,8 @@
import sqlmodel
from pynecone import utils
from pynecone.base import Base
from pynecone.config import get_config
def get_engine():
@ -15,7 +15,7 @@ def get_engine():
Raises:
ValueError: If the database url is None.
"""
url = utils.get_config().db_url
url = get_config().db_url
if url is None:
raise ValueError("No database url in config")
return sqlmodel.create_engine(url, echo=False)

92
pynecone/pc.py
View File

@ -7,8 +7,10 @@ from pathlib import Path
import httpx
import typer
from pynecone import constants, utils
from pynecone import constants
from pynecone.config import get_config
from pynecone.telemetry import pynecone_telemetry
from pynecone.utils import build, console, exec, prerequisites, processes
# Create the app.
cli = typer.Typer()
@ -17,40 +19,40 @@ cli = typer.Typer()
@cli.command()
def version():
"""Get the Pynecone version."""
utils.console.print(constants.VERSION)
console.print(constants.VERSION)
@cli.command()
def init():
"""Initialize a new Pynecone app in the current directory."""
app_name = utils.get_default_app_name()
app_name = prerequisites.get_default_app_name()
# Make sure they don't name the app "pynecone".
if app_name == constants.MODULE_NAME:
utils.console.print(
console.print(
f"[red]The app directory cannot be named [bold]{constants.MODULE_NAME}."
)
raise typer.Exit()
with utils.console.status(f"[bold]Initializing {app_name}"):
with console.status(f"[bold]Initializing {app_name}"):
# Set up the web directory.
utils.install_bun()
utils.initialize_web_directory()
prerequisites.install_bun()
prerequisites.initialize_web_directory()
# Set up the app directory, only if the config doesn't exist.
if not os.path.exists(constants.CONFIG_FILE):
utils.create_config(app_name)
utils.initialize_app_directory(app_name)
utils.set_pynecone_project_hash()
pynecone_telemetry("init", utils.get_config().telemetry_enabled)
prerequisites.create_config(app_name)
prerequisites.initialize_app_directory(app_name)
build.set_pynecone_project_hash()
pynecone_telemetry("init", get_config().telemetry_enabled)
else:
utils.set_pynecone_project_hash()
pynecone_telemetry("reinit", utils.get_config().telemetry_enabled)
build.set_pynecone_project_hash()
pynecone_telemetry("reinit", get_config().telemetry_enabled)
# Initialize the .gitignore.
utils.initialize_gitignore()
prerequisites.initialize_gitignore()
# Finish initializing the app.
utils.console.log(f"[bold green]Finished Initializing: {app_name}")
console.log(f"[bold green]Finished Initializing: {app_name}")
@cli.command()
@ -71,48 +73,48 @@ def run(
):
"""Run the app in the current directory."""
if platform.system() == "Windows":
utils.console.print(
console.print(
"[yellow][WARNING] We strongly advise you to use Windows Subsystem for Linux (WSL) for optimal performance when using Pynecone. Due to compatibility issues with one of our dependencies, Bun, you may experience slower performance on Windows. By using WSL, you can expect to see a significant speed increase."
)
frontend_port = utils.get_config().port if port is None else port
backend_port = utils.get_config().backend_port
frontend_port = get_config().port if port is None else port
backend_port = get_config().backend_port
# If something is running on the ports, ask the user if they want to kill or change it.
if utils.is_process_on_port(frontend_port):
frontend_port = utils.change_or_terminate_port(frontend_port, "frontend")
if processes.is_process_on_port(frontend_port):
frontend_port = processes.change_or_terminate_port(frontend_port, "frontend")
if utils.is_process_on_port(backend_port):
backend_port = utils.change_or_terminate_port(backend_port, "backend")
if processes.is_process_on_port(backend_port):
backend_port = processes.change_or_terminate_port(backend_port, "backend")
# Check that the app is initialized.
if frontend and not utils.is_initialized():
utils.console.print(
if frontend and not prerequisites.is_initialized():
console.print(
"[red]The app is not initialized. Run [bold]pc init[/bold] first."
)
raise typer.Exit()
# Check that the template is up to date.
if frontend and not utils.is_latest_template():
utils.console.print(
if frontend and not prerequisites.is_latest_template():
console.print(
"[red]The base app template has updated. Run [bold]pc init[/bold] again."
)
raise typer.Exit()
# Get the app module.
utils.console.rule("[bold]Starting Pynecone App")
app = utils.get_app()
console.rule("[bold]Starting Pynecone App")
app = prerequisites.get_app()
# Get the frontend and backend commands, based on the environment.
frontend_cmd = backend_cmd = None
if env == constants.Env.DEV:
frontend_cmd, backend_cmd = utils.run_frontend, utils.run_backend
frontend_cmd, backend_cmd = exec.run_frontend, exec.run_backend
if env == constants.Env.PROD:
frontend_cmd, backend_cmd = utils.run_frontend_prod, utils.run_backend_prod
frontend_cmd, backend_cmd = exec.run_frontend_prod, exec.run_backend_prod
assert frontend_cmd and backend_cmd, "Invalid env"
# Post a telemetry event.
pynecone_telemetry(f"run-{env.value}", utils.get_config().telemetry_enabled)
pynecone_telemetry(f"run-{env.value}", get_config().telemetry_enabled)
# Run the frontend and backend.
try:
@ -121,16 +123,16 @@ def run(
if backend:
backend_cmd(app.__name__, port=int(backend_port), loglevel=loglevel)
finally:
utils.kill_process_on_port(frontend_port)
utils.kill_process_on_port(backend_port)
processes.kill_process_on_port(frontend_port)
processes.kill_process_on_port(backend_port)
@cli.command()
def deploy(dry_run: bool = typer.Option(False, help="Whether to run a dry run.")):
"""Deploy the app to the Pynecone hosting service."""
# Get the app config.
config = utils.get_config()
config.api_url = utils.get_production_backend_url()
config = get_config()
config.api_url = prerequisites.get_production_backend_url()
# Check if the deploy url is set.
if config.pcdeploy_url is None:
@ -139,8 +141,8 @@ def deploy(dry_run: bool = typer.Option(False, help="Whether to run a dry run.")
# Compile the app in production mode.
typer.echo("Compiling production app")
app = utils.get_app().app
utils.export_app(app, zip=True, deploy_url=config.deploy_url)
app = prerequisites.get_app().app
build.export_app(app, zip=True, deploy_url=config.deploy_url)
# Exit early if this is a dry run.
if dry_run:
@ -179,16 +181,16 @@ def export(
),
):
"""Export the app to a zip file."""
config = utils.get_config()
config = get_config()
if for_pc_deploy:
# Get the app config and modify the api_url base on username and app_name.
config.api_url = utils.get_production_backend_url()
config.api_url = prerequisites.get_production_backend_url()
# Compile the app in production mode and export it.
utils.console.rule("[bold]Compiling production app and preparing for export.")
app = utils.get_app().app
utils.export_app(
console.rule("[bold]Compiling production app and preparing for export.")
app = prerequisites.get_app().app
build.export_app(
app,
backend=backend,
frontend=frontend,
@ -197,15 +199,15 @@ def export(
)
# Post a telemetry event.
pynecone_telemetry("export", utils.get_config().telemetry_enabled)
pynecone_telemetry("export", get_config().telemetry_enabled)
if zipping:
utils.console.rule(
console.rule(
"""Backend & Frontend compiled. See [green bold]backend.zip[/green bold]
and [green bold]frontend.zip[/green bold]."""
)
else:
utils.console.rule(
console.rule(
"""Backend & Frontend compiled. See [green bold]app[/green bold]
and [green bold].web/_static[/green bold] directories."""
)

100
pynecone/route.py
View File

@ -1,7 +1,11 @@
"""The route decorator and associated variables."""
"""The route decorator and associated variables and functions."""
from typing import List, Optional, Union
from __future__ import annotations
import re
from typing import Dict, List, Optional, Union
from pynecone import constants
from pynecone.event import EventHandler
DECORATED_ROUTES = []
@ -52,3 +56,95 @@ def route(
return render_fn
return decorator
def verify_route_validity(route: str) -> None:
"""Verify if the route is valid, and throw an error if not.
Args:
route: The route that need to be checked
Raises:
ValueError: If the route is invalid.
"""
pattern = catchall_in_route(route)
if pattern and not route.endswith(pattern):
raise ValueError(f"Catch-all must be the last part of the URL: {route}")
def get_route_args(route: str) -> Dict[str, str]:
"""Get the dynamic arguments for the given route.
Args:
route: The route to get the arguments for.
Returns:
The route arguments.
"""
args = {}
def add_route_arg(match: re.Match[str], type_: str):
"""Add arg from regex search result.
Args:
match: Result of a regex search
type_: The assigned type for this arg
Raises:
ValueError: If the route is invalid.
"""
arg_name = match.groups()[0]
if arg_name in args:
raise ValueError(
f"Arg name [{arg_name}] is used more than once in this URL"
)
args[arg_name] = type_
# Regex to check for route args.
check = constants.RouteRegex.ARG
check_strict_catchall = constants.RouteRegex.STRICT_CATCHALL
check_opt_catchall = constants.RouteRegex.OPT_CATCHALL
# Iterate over the route parts and check for route args.
for part in route.split("/"):
match_opt = check_opt_catchall.match(part)
if match_opt:
add_route_arg(match_opt, constants.RouteArgType.LIST)
break
match_strict = check_strict_catchall.match(part)
if match_strict:
add_route_arg(match_strict, constants.RouteArgType.LIST)
break
match = check.match(part)
if match:
# Add the route arg to the list.
add_route_arg(match, constants.RouteArgType.SINGLE)
return args
def catchall_in_route(route: str) -> str:
"""Extract the catchall part from a route.
Args:
route: the route from which to extract
Returns:
str: the catchall part of the URI
"""
match_ = constants.RouteRegex.CATCHALL.search(route)
return match_.group() if match_ else ""
def catchall_prefix(route: str) -> str:
"""Extract the prefix part from a route that contains a catchall.
Args:
route: the route from which to extract
Returns:
str: the prefix part of the URI
"""
pattern = catchall_in_route(route)
return route.replace(pattern, "") if pattern else ""

25
pynecone/state.py
View File

@ -21,9 +21,10 @@ from typing import (
import cloudpickle
from redis import Redis
from pynecone import constants, utils
from pynecone import constants
from pynecone.base import Base
from pynecone.event import Event, EventHandler, window_alert
from pynecone.event import Event, EventHandler, fix_events, window_alert
from pynecone.utils import format, prerequisites, types
from pynecone.var import BaseVar, ComputedVar, PCDict, PCList, Var
Delta = Dict[str, Any]
@ -93,7 +94,7 @@ class State(Base, ABC):
value, self._reassign_field, field.name
)
if utils._issubclass(field.type_, Union[List, Dict]):
if types._issubclass(field.type_, Union[List, Dict]):
setattr(self, field.name, value_in_pc_data)
self.clean()
@ -138,7 +139,7 @@ class State(Base, ABC):
cls.new_backend_vars = {
name: value
for name, value in cls.__dict__.items()
if utils.is_backend_variable(name)
if types.is_backend_variable(name)
and name not in cls.inherited_backend_vars
}
@ -193,7 +194,7 @@ class State(Base, ABC):
parent_states = [
base
for base in cls.__bases__
if utils._issubclass(base, State) and base is not State
if types._issubclass(base, State) and base is not State
]
assert len(parent_states) < 2, "Only one parent state is allowed."
return parent_states[0] if len(parent_states) == 1 else None # type: ignore
@ -216,7 +217,7 @@ class State(Base, ABC):
Returns:
The name of the state.
"""
return utils.to_snake_case(cls.__name__)
return format.to_snake_case(cls.__name__)
@classmethod
@functools.lru_cache()
@ -286,7 +287,7 @@ class State(Base, ABC):
Raises:
TypeError: if the variable has an incorrect type
"""
if not utils.is_valid_var_type(prop.type_):
if not types.is_valid_var_type(prop.type_):
raise TypeError(
"State vars must be primitive Python types, "
"Plotly figures, Pandas dataframes, "
@ -482,7 +483,7 @@ class State(Base, ABC):
setattr(self.parent_state, name, value)
return
if utils.is_backend_variable(name):
if types.is_backend_variable(name):
self.backend_vars.__setitem__(name, value)
self.mark_dirty()
return
@ -556,13 +557,13 @@ class State(Base, ABC):
except Exception:
error = traceback.format_exc()
print(error)
events = utils.fix_events(
events = fix_events(
[window_alert("An error occurred. See logs for details.")], event.token
)
return StateUpdate(events=events)
# Fix the returned events.
events = utils.fix_events(events, event.token)
events = fix_events(events, event.token)
# Get the delta after processing the event.
delta = self.get_delta()
@ -595,7 +596,7 @@ class State(Base, ABC):
delta.update(substates[substate].get_delta())
# Format the delta.
delta = utils.format_state(delta)
delta = format.format_state(delta)
# Return the delta.
return delta
@ -685,7 +686,7 @@ class StateManager(Base):
state: The state class to use.
"""
self.state = state
self.redis = utils.get_redis()
self.redis = prerequisites.get_redis()
def get_state(self, token: str) -> State:
"""Get the state for a token.

5
pynecone/style.py
View File

@ -2,8 +2,9 @@
from typing import Optional
from pynecone import constants, utils
from pynecone import constants
from pynecone.event import EventChain
from pynecone.utils import format
from pynecone.var import BaseVar, Var
toggle_color_mode = BaseVar(name=constants.TOGGLE_COLOR_MODE, type_=EventChain)
@ -20,7 +21,7 @@ def convert(style_dict):
"""
out = {}
for key, value in style_dict.items():
key = utils.to_camel_case(key)
key = format.to_camel_case(key)
if isinstance(value, dict):
out[key] = convert(value)
elif isinstance(value, Var):

1606
pynecone/utils.py
View File

File diff suppressed because it is too large Load Diff

147
pynecone/utils/build.py Normal file
View File

@ -0,0 +1,147 @@
"""Building the app and initializing all prerequisites."""
from __future__ import annotations
import json
import os
import random
import subprocess
from pathlib import Path
from typing import (
TYPE_CHECKING,
Optional,
)
from pynecone import constants
from pynecone.config import get_config
from pynecone.utils import path_ops, prerequisites
if TYPE_CHECKING:
from pynecone.app import App
def set_pynecone_project_hash():
"""Write the hash of the Pynecone project to a PCVERSION_APP_FILE."""
with open(constants.PCVERSION_APP_FILE) as f: # type: ignore
pynecone_json = json.load(f)
pynecone_json["project_hash"] = random.getrandbits(128)
with open(constants.PCVERSION_APP_FILE, "w") as f:
json.dump(pynecone_json, f, ensure_ascii=False)
def generate_sitemap(deploy_url: str):
"""Generate the sitemap config file.
Args:
deploy_url: The URL of the deployed app.
"""
# Import here to avoid circular imports.
from pynecone.compiler import templates
config = json.dumps(
{
"siteUrl": deploy_url,
"generateRobotsTxt": True,
}
)
with open(constants.SITEMAP_CONFIG_FILE, "w") as f:
f.write(templates.SITEMAP_CONFIG(config=config))
def export_app(
app: App,
backend: bool = True,
frontend: bool = True,
zip: bool = False,
deploy_url: Optional[str] = None,
):
"""Zip up the app for deployment.
Args:
app: The app.
backend: Whether to zip up the backend app.
frontend: Whether to zip up the frontend app.
zip: Whether to zip the app.
deploy_url: The URL of the deployed app.
"""
# Force compile the app.
app.compile(force_compile=True)
# Remove the static folder.
path_ops.rm(constants.WEB_STATIC_DIR)
# Generate the sitemap file.
if deploy_url is not None:
generate_sitemap(deploy_url)
# Export the Next app.
subprocess.run(
[prerequisites.get_package_manager(), "run", "export"], cwd=constants.WEB_DIR
)
# Zip up the app.
if zip:
if os.name == "posix":
posix_export(backend, frontend)
if os.name == "nt":
nt_export(backend, frontend)
def nt_export(backend: bool = True, frontend: bool = True):
"""Export for nt (Windows) systems.
Args:
backend: Whether to zip up the backend app.
frontend: Whether to zip up the frontend app.
"""
cmd = r""
if frontend:
cmd = r'''powershell -Command "Set-Location .web/_static; Compress-Archive -Path .\* -DestinationPath ..\..\frontend.zip -Force"'''
os.system(cmd)
if backend:
cmd = r'''powershell -Command "Get-ChildItem -File | Where-Object { $_.Name -notin @('.web', 'assets', 'frontend.zip', 'backend.zip') } | Compress-Archive -DestinationPath backend.zip -Update"'''
os.system(cmd)
def posix_export(backend: bool = True, frontend: bool = True):
"""Export for posix (Linux, OSX) systems.
Args:
backend: Whether to zip up the backend app.
frontend: Whether to zip up the frontend app.
"""
cmd = r""
if frontend:
cmd = r"cd .web/_static && zip -r ../../frontend.zip ./*"
os.system(cmd)
if backend:
cmd = r"zip -r backend.zip ./* -x .web/\* ./assets\* ./frontend.zip\* ./backend.zip\*"
os.system(cmd)
def setup_frontend(root: Path):
"""Set up the frontend.
Args:
root: root path of the project.
"""
# copy asset files to public folder
path_ops.mkdir(str(root / constants.WEB_ASSETS_DIR))
path_ops.cp(
src=str(root / constants.APP_ASSETS_DIR),
dest=str(root / constants.WEB_ASSETS_DIR),
)
def setup_backend():
"""Set up backend.
Specifically ensures backend database is updated when running --no-frontend.
"""
# Import here to avoid circular imports.
from pynecone.model import Model
config = get_config()
if config.db_url is not None:
Model.create_all()

75
pynecone/utils/console.py Normal file
View File

@ -0,0 +1,75 @@
"""Functions to communicate to the user via console."""
from __future__ import annotations
from typing import List, Optional
from rich.console import Console
from rich.prompt import Prompt
from rich.status import Status
# Console for pretty printing.
_console = Console()
def deprecate(msg: str) -> None:
"""Print a deprecation warning.
Args:
msg: The deprecation message.
"""
_console.print(f"[yellow]DeprecationWarning: {msg}[/yellow]")
def log(msg: str) -> None:
"""Takes a string and logs it to the console.
Args:
msg (str): The message to log.
"""
_console.log(msg)
def print(msg: str) -> None:
"""Prints the given message to the console.
Args:
msg (str): The message to print to the console.
"""
_console.print(msg)
def rule(title: str) -> None:
"""Prints a horizontal rule with a title.
Args:
title (str): The title of the rule.
"""
_console.rule(title)
def ask(question: str, choices: Optional[List[str]] = None) -> str:
"""Takes a prompt question and optionally a list of choices
and returns the user input.
Args:
question (str): The question to ask the user.
choices (Optional[List[str]]): A list of choices to select from
Returns:
A string
"""
return Prompt.ask(question, choices=choices)
def status(msg: str) -> Status:
"""Returns a status,
which can be used as a context manager.
Args:
msg (str): The message to be used as status title.
Returns:
The status of the console.
"""
return _console.status(msg)

159
pynecone/utils/exec.py Normal file
View File

@ -0,0 +1,159 @@
"""Everything regarding execution of the built app."""
from __future__ import annotations
import os
import platform
import subprocess
from pathlib import Path
from typing import TYPE_CHECKING
import uvicorn
from pynecone import constants
from pynecone.config import get_config
from pynecone.utils import console, prerequisites, processes
from pynecone.utils.build import export_app, setup_backend, setup_frontend
from pynecone.watch import AssetFolderWatch
if TYPE_CHECKING:
from pynecone.app import App
def start_watching_assets_folder(root):
"""Start watching assets folder.
Args:
root: root path of the project.
"""
asset_watch = AssetFolderWatch(root)
asset_watch.start()
def run_frontend(app: App, root: Path, port: str):
"""Run the frontend.
Args:
app: The app.
root: root path of the project.
port: port of the app.
"""
# Initialize the web directory if it doesn't exist.
web_dir = prerequisites.create_web_directory(root)
# Install frontend packages
prerequisites.install_frontend_packages(web_dir)
# Set up the frontend.
setup_frontend(root)
# start watching asset folder
start_watching_assets_folder(root)
# Compile the frontend.
app.compile(force_compile=True)
# Run the frontend in development mode.
console.rule("[bold green]App Running")
os.environ["PORT"] = get_config().port if port is None else port
subprocess.Popen(
[prerequisites.get_package_manager(), "run", "next", "telemetry", "disable"],
cwd=constants.WEB_DIR,
env=os.environ,
stdout=subprocess.DEVNULL,
stderr=subprocess.STDOUT,
)
subprocess.Popen(
[prerequisites.get_package_manager(), "run", "dev"],
cwd=constants.WEB_DIR,
env=os.environ,
)
def run_frontend_prod(app: App, root: Path, port: str):
"""Run the frontend.
Args:
app: The app.
root: root path of the project.
port: port of the app.
"""
# Set up the frontend.
setup_frontend(root)
# Export the app.
export_app(app)
os.environ["PORT"] = get_config().port if port is None else port
# Run the frontend in production mode.
subprocess.Popen(
[prerequisites.get_package_manager(), "run", "prod"],
cwd=constants.WEB_DIR,
env=os.environ,
)
def run_backend(
app_name: str, port: int, loglevel: constants.LogLevel = constants.LogLevel.ERROR
):
"""Run the backend.
Args:
app_name: The app name.
port: The app port
loglevel: The log level.
"""
setup_backend()
uvicorn.run(
f"{app_name}:{constants.APP_VAR}.{constants.API_VAR}",
host=constants.BACKEND_HOST,
port=port,
log_level=loglevel,
reload=True,
)
def run_backend_prod(
app_name: str, port: int, loglevel: constants.LogLevel = constants.LogLevel.ERROR
):
"""Run the backend.
Args:
app_name: The app name.
port: The app port
loglevel: The log level.
"""
setup_backend()
num_workers = processes.get_num_workers()
command = (
[
*constants.RUN_BACKEND_PROD_WINDOWS,
"--host",
"0.0.0.0",
"--port",
str(port),
f"{app_name}:{constants.APP_VAR}",
]
if platform.system() == "Windows"
else [
*constants.RUN_BACKEND_PROD,
"--bind",
f"0.0.0.0:{port}",
"--threads",
str(num_workers),
f"{app_name}:{constants.APP_VAR}()",
]
)
command += [
"--log-level",
loglevel.value,
"--workers",
str(num_workers),
]
subprocess.run(command)

387
pynecone/utils/format.py Normal file
View File

@ -0,0 +1,387 @@
"""Formatting operations."""
from __future__ import annotations
import json
import os
import re
import sys
from typing import TYPE_CHECKING, Any, Dict, List, Optional, Tuple, Type
import plotly.graph_objects as go
from plotly.io import to_json
from pynecone import constants
from pynecone.utils import types
if TYPE_CHECKING:
from pynecone.event import EventHandler, EventSpec
WRAP_MAP = {
"{": "}",
"(": ")",
"[": "]",
"<": ">",
'"': '"',
"'": "'",
"`": "`",
}
def get_close_char(open: str, close: Optional[str] = None) -> str:
"""Check if the given character is a valid brace.
Args:
open: The open character.
close: The close character if provided.
Returns:
The close character.
Raises:
ValueError: If the open character is not a valid brace.
"""
if close is not None:
return close
if open not in WRAP_MAP:
raise ValueError(f"Invalid wrap open: {open}, must be one of {WRAP_MAP.keys()}")
return WRAP_MAP[open]
def is_wrapped(text: str, open: str, close: Optional[str] = None) -> bool:
"""Check if the given text is wrapped in the given open and close characters.
Args:
text: The text to check.
open: The open character.
close: The close character.
Returns:
Whether the text is wrapped.
"""
close = get_close_char(open, close)
return text.startswith(open) and text.endswith(close)
def wrap(
text: str,
open: str,
close: Optional[str] = None,
check_first: bool = True,
num: int = 1,
) -> str:
"""Wrap the given text in the given open and close characters.
Args:
text: The text to wrap.
open: The open character.
close: The close character.
check_first: Whether to check if the text is already wrapped.
num: The number of times to wrap the text.
Returns:
The wrapped text.
"""
close = get_close_char(open, close)
# If desired, check if the text is already wrapped in braces.
if check_first and is_wrapped(text=text, open=open, close=close):
return text
# Wrap the text in braces.
return f"{open * num}{text}{close * num}"
def indent(text: str, indent_level: int = 2) -> str:
"""Indent the given text by the given indent level.
Args:
text: The text to indent.
indent_level: The indent level.
Returns:
The indented text.
"""
lines = text.splitlines()
if len(lines) < 2:
return text
return os.linesep.join(f"{' ' * indent_level}{line}" for line in lines) + os.linesep
def to_snake_case(text: str) -> str:
"""Convert a string to snake case.
The words in the text are converted to lowercase and
separated by underscores.
Args:
text: The string to convert.
Returns:
The snake case string.
"""
s1 = re.sub("(.)([A-Z][a-z]+)", r"\1_\2", text)
return re.sub("([a-z0-9])([A-Z])", r"\1_\2", s1).lower()
def to_camel_case(text: str) -> str:
"""Convert a string to camel case.
The first word in the text is converted to lowercase and
the rest of the words are converted to title case, removing underscores.
Args:
text: The string to convert.
Returns:
The camel case string.
"""
if "_" not in text:
return text
camel = "".join(
word.capitalize() if i > 0 else word.lower()
for i, word in enumerate(text.lstrip("_").split("_"))
)
prefix = "_" if text.startswith("_") else ""
return prefix + camel
def to_title_case(text: str) -> str:
"""Convert a string from snake case to title case.
Args:
text: The string to convert.
Returns:
The title case string.
"""
return "".join(word.capitalize() for word in text.split("_"))
def format_string(string: str) -> str:
"""Format the given string as a JS string literal..
Args:
string: The string to format.
Returns:
The formatted string.
"""
# Escape backticks.
string = string.replace(r"\`", "`")
string = string.replace("`", r"\`")
# Wrap the string so it looks like {`string`}.
string = wrap(string, "`")
string = wrap(string, "{")
return string
def format_route(route: str) -> str:
"""Format the given route.
Args:
route: The route to format.
Returns:
The formatted route.
"""
# Strip the route.
route = route.strip("/")
route = to_snake_case(route).replace("_", "-")
# If the route is empty, return the index route.
if route == "":
return constants.INDEX_ROUTE
return route
def format_cond(
cond: str,
true_value: str,
false_value: str = '""',
is_prop=False,
) -> str:
"""Format a conditional expression.
Args:
cond: The cond.
true_value: The value to return if the cond is true.
false_value: The value to return if the cond is false.
is_prop: Whether the cond is a prop
Returns:
The formatted conditional expression.
"""
# Import here to avoid circular imports.
from pynecone.var import Var
# Format prop conds.
if is_prop:
prop1 = Var.create(true_value, is_string=type(true_value) == str)
prop2 = Var.create(false_value, is_string=type(false_value) == str)
assert prop1 is not None and prop2 is not None, "Invalid prop values"
return f"{cond} ? {prop1} : {prop2}".replace("{", "").replace("}", "")
# Format component conds.
return wrap(f"{cond} ? {true_value} : {false_value}", "{")
def get_event_handler_parts(handler: EventHandler) -> Tuple[str, str]:
"""Get the state and function name of an event handler.
Args:
handler: The event handler to get the parts of.
Returns:
The state and function name.
"""
# Get the class that defines the event handler.
parts = handler.fn.__qualname__.split(".")
# If there's no enclosing class, just return the function name.
if len(parts) == 1:
return ("", parts[-1])
# Get the state and the function name.
state_name, name = parts[-2:]
# Construct the full event handler name.
try:
# Try to get the state from the module.
state = vars(sys.modules[handler.fn.__module__])[state_name]
except Exception:
# If the state isn't in the module, just return the function name.
return ("", handler.fn.__qualname__)
return (state.get_full_name(), name)
def format_event_handler(handler: EventHandler) -> str:
"""Format an event handler.
Args:
handler: The event handler to format.
Returns:
The formatted function.
"""
state, name = get_event_handler_parts(handler)
if state == "":
return name
return f"{state}.{name}"
def format_event(event_spec: EventSpec) -> str:
"""Format an event.
Args:
event_spec: The event to format.
Returns:
The compiled event.
"""
args = ",".join([":".join((name, val)) for name, val in event_spec.args])
return f"E(\"{format_event_handler(event_spec.handler)}\", {wrap(args, '{')})"
def format_upload_event(event_spec: EventSpec) -> str:
"""Format an upload event.
Args:
event_spec: The event to format.
Returns:
The compiled event.
"""
from pynecone.compiler import templates
state, name = get_event_handler_parts(event_spec.handler)
return f'uploadFiles({state}, {templates.RESULT}, {templates.SET_RESULT}, {state}.files, "{name}", UPLOAD)'
def format_query_params(router_data: Dict[str, Any]) -> Dict[str, str]:
"""Convert back query params name to python-friendly case.
Args:
router_data: the router_data dict containing the query params
Returns:
The reformatted query params
"""
params = router_data[constants.RouteVar.QUERY]
return {k.replace("-", "_"): v for k, v in params.items()}
def format_dataframe_values(value: Type) -> List[Any]:
"""Format dataframe values.
Args:
value: The value to format.
Returns:
Format data
"""
if not types.is_dataframe(type(value)):
return value
format_data = []
for data in list(value.values.tolist()):
element = []
for d in data:
element.append(str(d) if isinstance(d, (list, tuple)) else d)
format_data.append(element)
return format_data
def format_state(value: Any) -> Dict:
"""Recursively format values in the given state.
Args:
value: The state to format.
Returns:
The formatted state.
Raises:
TypeError: If the given value is not a valid state.
"""
# Handle dicts.
if isinstance(value, dict):
return {k: format_state(v) for k, v in value.items()}
# Return state vars as is.
if isinstance(value, types.StateBases):
return value
# Convert plotly figures to JSON.
if isinstance(value, go.Figure):
return json.loads(to_json(value))["data"] # type: ignore
# Convert pandas dataframes to JSON.
if types.is_dataframe(type(value)):
return {
"columns": value.columns.tolist(),
"data": format_dataframe_values(value),
}
raise TypeError(
"State vars must be primitive Python types, "
"or subclasses of pc.Base. "
f"Got var of type {type(value)}."
)
def json_dumps(obj: Any) -> str:
"""Takes an object and returns a jsonified string.
Args:
obj: The object to be serialized.
Returns:
A string
"""
return json.dumps(obj, ensure_ascii=False)

23
pynecone/utils/imports.py Normal file
View File

@ -0,0 +1,23 @@
"""Import operations."""
from collections import defaultdict
from typing import Dict, Set
ImportDict = Dict[str, Set[str]]
def merge_imports(*imports) -> ImportDict:
"""Merge two import dicts together.
Args:
*imports: The list of import dicts to merge.
Returns:
The merged import dicts.
"""
all_imports = defaultdict(set)
for import_dict in imports:
for lib, fields in import_dict.items():
for field in fields:
all_imports[lib].add(field)
return all_imports

110
pynecone/utils/path_ops.py Normal file
View File

@ -0,0 +1,110 @@
"""Path operations."""
from __future__ import annotations
import os
import shutil
from typing import Optional
# Shorthand for join.
join = os.linesep.join
def rm(path: str):
"""Remove a file or directory.
Args:
path: The path to the file or directory.
"""
if os.path.isdir(path):
shutil.rmtree(path)
elif os.path.isfile(path):
os.remove(path)
def cp(src: str, dest: str, overwrite: bool = True) -> bool:
"""Copy a file or directory.
Args:
src: The path to the file or directory.
dest: The path to the destination.
overwrite: Whether to overwrite the destination.
Returns:
Whether the copy was successful.
"""
if src == dest:
return False
if not overwrite and os.path.exists(dest):
return False
if os.path.isdir(src):
rm(dest)
shutil.copytree(src, dest)
else:
shutil.copyfile(src, dest)
return True
def mv(src: str, dest: str, overwrite: bool = True) -> bool:
"""Move a file or directory.
Args:
src: The path to the file or directory.
dest: The path to the destination.
overwrite: Whether to overwrite the destination.
Returns:
Whether the move was successful.
"""
if src == dest:
return False
if not overwrite and os.path.exists(dest):
return False
rm(dest)
shutil.move(src, dest)
return True
def mkdir(path: str):
"""Create a directory.
Args:
path: The path to the directory.
"""
if not os.path.exists(path):
os.makedirs(path)
def ln(src: str, dest: str, overwrite: bool = False) -> bool:
"""Create a symbolic link.
Args:
src: The path to the file or directory.
dest: The path to the destination.
overwrite: Whether to overwrite the destination.
Returns:
Whether the link was successful.
"""
if src == dest:
return False
if not overwrite and (os.path.exists(dest) or os.path.islink(dest)):
return False
if os.path.isdir(src):
rm(dest)
os.symlink(src, dest, target_is_directory=True)
else:
os.symlink(src, dest)
return True
def which(program: str) -> Optional[str]:
"""Find the path to an executable.
Args:
program: The name of the executable.
Returns:
The path to the executable.
"""
return shutil.which(program)

265
pynecone/utils/prerequisites.py Normal file
View File

@ -0,0 +1,265 @@
"""Everything related to fetching or initializing build prerequisites."""
from __future__ import annotations
import json
import os
import platform
import subprocess
import sys
from pathlib import Path
from types import ModuleType
from typing import Optional
import typer
from redis import Redis
from pynecone import constants
from pynecone.config import get_config
from pynecone.utils import console, path_ops
def check_node_version(min_version):
"""Check the version of Node.js.
Args:
min_version: The minimum version of Node.js required.
Returns:
Whether the version of Node.js is high enough.
"""
try:
# Run the node -v command and capture the output
result = subprocess.run(
["node", "-v"], stdout=subprocess.PIPE, stderr=subprocess.PIPE
)
# The output will be in the form "vX.Y.Z", so we can split it on the "v" character and take the second part
version = result.stdout.decode().strip().split("v")[1]
# Compare the version numbers
return version.split(".") >= min_version.split(".")
except Exception:
return False
def get_package_manager() -> str:
"""Get the package manager executable.
Returns:
The path to the package manager.
Raises:
FileNotFoundError: If bun or npm is not installed.
Exit: If the app directory is invalid.
"""
# Check that the node version is valid.
if not check_node_version(constants.MIN_NODE_VERSION):
console.print(
f"[red]Node.js version {constants.MIN_NODE_VERSION} or higher is required to run Pynecone."
)
raise typer.Exit()
# On Windows, we use npm instead of bun.
if platform.system() == "Windows":
npm_path = path_ops.which("npm")
if npm_path is None:
raise FileNotFoundError("Pynecone requires npm to be installed on Windows.")
return npm_path
# On other platforms, we use bun.
return os.path.expandvars(get_config().bun_path)
def get_app() -> ModuleType:
"""Get the app module based on the default config.
Returns:
The app based on the default config.
"""
config = get_config()
module = ".".join([config.app_name, config.app_name])
sys.path.insert(0, os.getcwd())
return __import__(module, fromlist=(constants.APP_VAR,))
def get_redis() -> Optional[Redis]:
"""Get the redis client.
Returns:
The redis client.
"""
config = get_config()
if config.redis_url is None:
return None
redis_url, redis_port = config.redis_url.split(":")
print("Using redis at", config.redis_url)
return Redis(host=redis_url, port=int(redis_port), db=0)
def get_production_backend_url() -> str:
"""Get the production backend URL.
Returns:
The production backend URL.
"""
config = get_config()
return constants.PRODUCTION_BACKEND_URL.format(
username=config.username,
app_name=config.app_name,
)
def get_default_app_name() -> str:
"""Get the default app name.
The default app name is the name of the current directory.
Returns:
The default app name.
"""
return os.getcwd().split(os.path.sep)[-1].replace("-", "_")
def create_config(app_name: str):
"""Create a new pcconfig file.
Args:
app_name: The name of the app.
"""
# Import here to avoid circular imports.
from pynecone.compiler import templates
with open(constants.CONFIG_FILE, "w") as f:
f.write(templates.PCCONFIG.format(app_name=app_name))
def create_web_directory(root: Path) -> str:
"""Creates a web directory in the given root directory
and returns the path to the directory.
Args:
root (Path): The root directory of the project.
Returns:
The path to the web directory.
"""
web_dir = str(root / constants.WEB_DIR)
path_ops.cp(constants.WEB_TEMPLATE_DIR, web_dir, overwrite=False)
return web_dir
def initialize_gitignore():
"""Initialize the template .gitignore file."""
# The files to add to the .gitignore file.
files = constants.DEFAULT_GITIGNORE
# Subtract current ignored files.
if os.path.exists(constants.GITIGNORE_FILE):
with open(constants.GITIGNORE_FILE, "r") as f:
files -= set(f.read().splitlines())
# Add the new files to the .gitignore file.
with open(constants.GITIGNORE_FILE, "a") as f:
f.write(path_ops.join(files))
def initialize_app_directory(app_name: str):
"""Initialize the app directory on pc init.
Args:
app_name: The name of the app.
"""
console.log("Initializing the app directory.")
path_ops.cp(constants.APP_TEMPLATE_DIR, app_name)
path_ops.mv(
os.path.join(app_name, constants.APP_TEMPLATE_FILE),
os.path.join(app_name, app_name + constants.PY_EXT),
)
path_ops.cp(constants.ASSETS_TEMPLATE_DIR, constants.APP_ASSETS_DIR)
def initialize_web_directory():
"""Initialize the web directory on pc init."""
console.log("Initializing the web directory.")
path_ops.rm(os.path.join(constants.WEB_TEMPLATE_DIR, constants.NODE_MODULES))
path_ops.rm(os.path.join(constants.WEB_TEMPLATE_DIR, constants.PACKAGE_LOCK))
path_ops.cp(constants.WEB_TEMPLATE_DIR, constants.WEB_DIR)
def install_bun():
"""Install bun onto the user's system.
Raises:
FileNotFoundError: If the required packages are not installed.
"""
# Bun is not supported on Windows.
if platform.system() == "Windows":
console.log("Skipping bun installation on Windows.")
return
# Only install if bun is not already installed.
if not os.path.exists(get_package_manager()):
console.log("Installing bun...")
# Check if curl is installed
curl_path = path_ops.which("curl")
if curl_path is None:
raise FileNotFoundError("Pynecone requires curl to be installed.")
# Check if unzip is installed
unzip_path = path_ops.which("unzip")
if unzip_path is None:
raise FileNotFoundError("Pynecone requires unzip to be installed.")
os.system(constants.INSTALL_BUN)
def install_frontend_packages(web_dir: str):
"""Installs the base and custom frontend packages
into the given web directory.
Args:
web_dir (str): The directory where the frontend code is located.
"""
# Install the frontend packages.
console.rule("[bold]Installing frontend packages")
# Install the base packages.
subprocess.run(
[get_package_manager(), "install"],
cwd=web_dir,
stdout=subprocess.PIPE,
)
# Install the app packages.
packages = get_config().frontend_packages
if len(packages) > 0:
subprocess.run(
[get_package_manager(), "add", *packages],
cwd=web_dir,
stdout=subprocess.PIPE,
)
def is_initialized() -> bool:
"""Check whether the app is initialized.
Returns:
Whether the app is initialized in the current directory.
"""
return os.path.exists(constants.CONFIG_FILE) and os.path.exists(constants.WEB_DIR)
def is_latest_template() -> bool:
"""Whether the app is using the latest template.
Returns:
Whether the app is using the latest template.
"""
with open(constants.PCVERSION_TEMPLATE_FILE) as f: # type: ignore
template_version = json.load(f)["version"]
if not os.path.exists(constants.PCVERSION_APP_FILE):
return False
with open(constants.PCVERSION_APP_FILE) as f: # type: ignore
app_version = json.load(f)["version"]
return app_version >= template_version

122
pynecone/utils/processes.py Normal file
View File

@ -0,0 +1,122 @@
"""Process operations."""
from __future__ import annotations
import contextlib
import os
import signal
import sys
from typing import Optional
from urllib.parse import urlparse
import psutil
from pynecone import constants
from pynecone.config import get_config
from pynecone.utils import console, prerequisites
def kill(pid):
"""Kill a process.
Args:
pid: The process ID.
"""
os.kill(pid, signal.SIGTERM)
def get_num_workers() -> int:
"""Get the number of backend worker processes.
Returns:
The number of backend worker processes.
"""
return 1 if prerequisites.get_redis() is None else (os.cpu_count() or 1) * 2 + 1
def get_api_port() -> int:
"""Get the API port.
Returns:
The API port.
"""
port = urlparse(get_config().api_url).port
if port is None:
port = urlparse(constants.API_URL).port
assert port is not None
return port
def get_process_on_port(port) -> Optional[psutil.Process]:
"""Get the process on the given port.
Args:
port: The port.
Returns:
The process on the given port.
"""
for proc in psutil.process_iter(["pid", "name", "cmdline"]):
try:
for conns in proc.connections(kind="inet"):
if conns.laddr.port == int(port):
return proc
except (psutil.NoSuchProcess, psutil.AccessDenied, psutil.ZombieProcess):
pass
return None
def is_process_on_port(port) -> bool:
"""Check if a process is running on the given port.
Args:
port: The port.
Returns:
Whether a process is running on the given port.
"""
return get_process_on_port(port) is not None
def kill_process_on_port(port):
"""Kill the process on the given port.
Args:
port: The port.
"""
if get_process_on_port(port) is not None:
with contextlib.suppress(psutil.AccessDenied):
get_process_on_port(port).kill() # type: ignore
def change_or_terminate_port(port, _type) -> str:
"""Terminate or change the port.
Args:
port: The port.
_type: The type of the port.
Returns:
The new port or the current one.
"""
console.print(
f"Something is already running on port [bold underline]{port}[/bold underline]. This is the port the {_type} runs on."
)
frontend_action = console.ask("Kill or change it?", choices=["k", "c", "n"])
if frontend_action == "k":
kill_process_on_port(port)
return port
elif frontend_action == "c":
new_port = console.ask("Specify the new port")
# Check if also the new port is used
if is_process_on_port(new_port):
return change_or_terminate_port(new_port, _type)
else:
console.print(
f"The {_type} will run on port [bold underline]{new_port}[/bold underline]."
)
return new_port
else:
console.print("Exiting...")
sys.exit()

178
pynecone/utils/types.py Normal file
View File

@ -0,0 +1,178 @@
"""Contains custom types and methods to check types."""
from __future__ import annotations
import contextlib
from typing import Any, Callable, Tuple, Type, Union, _GenericAlias # type: ignore
from pynecone.base import Base
# Union of generic types.
GenericType = Union[Type, _GenericAlias]
# Valid state var types.
PrimitiveType = Union[int, float, bool, str, list, dict, tuple]
StateVar = Union[PrimitiveType, Base, None]
def get_args(alias: _GenericAlias) -> Tuple[Type, ...]:
"""Get the arguments of a type alias.
Args:
alias: The type alias.
Returns:
The arguments of the type alias.
"""
return alias.__args__
def is_generic_alias(cls: GenericType) -> bool:
"""Check whether the class is a generic alias.
Args:
cls: The class to check.
Returns:
Whether the class is a generic alias.
"""
# For older versions of Python.
if isinstance(cls, _GenericAlias):
return True
with contextlib.suppress(ImportError):
from typing import _SpecialGenericAlias # type: ignore
if isinstance(cls, _SpecialGenericAlias):
return True
# For newer versions of Python.
try:
from types import GenericAlias # type: ignore
return isinstance(cls, GenericAlias)
except ImportError:
return False
def is_union(cls: GenericType) -> bool:
"""Check if a class is a Union.
Args:
cls: The class to check.
Returns:
Whether the class is a Union.
"""
with contextlib.suppress(ImportError):
from typing import _UnionGenericAlias # type: ignore
return isinstance(cls, _UnionGenericAlias)
return cls.__origin__ == Union if is_generic_alias(cls) else False
def get_base_class(cls: GenericType) -> Type:
"""Get the base class of a class.
Args:
cls: The class.
Returns:
The base class of the class.
"""
if is_union(cls):
return tuple(get_base_class(arg) for arg in get_args(cls))
return get_base_class(cls.__origin__) if is_generic_alias(cls) else cls
def _issubclass(cls: GenericType, cls_check: GenericType) -> bool:
"""Check if a class is a subclass of another class.
Args:
cls: The class to check.
cls_check: The class to check against.
Returns:
Whether the class is a subclass of the other class.
"""
# Special check for Any.
if cls_check == Any:
return True
if cls in [Any, Callable]:
return False
# Get the base classes.
cls_base = get_base_class(cls)
cls_check_base = get_base_class(cls_check)
# The class we're checking should not be a union.
if isinstance(cls_base, tuple):
return False
# Check if the types match.
return cls_check_base == Any or issubclass(cls_base, cls_check_base)
def _isinstance(obj: Any, cls: GenericType) -> bool:
"""Check if an object is an instance of a class.
Args:
obj: The object to check.
cls: The class to check against.
Returns:
Whether the object is an instance of the class.
"""
return isinstance(obj, get_base_class(cls))
def is_dataframe(value: Type) -> bool:
"""Check if the given value is a dataframe.
Args:
value: The value to check.
Returns:
Whether the value is a dataframe.
"""
return value.__name__ == "DataFrame"
def is_figure(value: Type) -> bool:
"""Check if the given value is a figure.
Args:
value: The value to check.
Returns:
Whether the value is a figure.
"""
return value.__name__ == "Figure"
def is_valid_var_type(var: Type) -> bool:
"""Check if the given value is a valid prop type.
Args:
var: The value to check.
Returns:
Whether the value is a valid prop type.
"""
return _issubclass(var, StateVar) or is_dataframe(var) or is_figure(var)
def is_backend_variable(name: str) -> bool:
"""Check if this variable name correspond to a backend variable.
Args:
name: The name of the variable to check
Returns:
bool: The result of the check
"""
return name.startswith("_") and not name.startswith("__")
# Store this here for performance.
StateBases = get_base_class(StateVar)

67
pynecone/var.py
View File

@ -2,6 +2,8 @@
from __future__ import annotations
import json
import random
import string
from abc import ABC
from typing import (
TYPE_CHECKING,
@ -19,13 +21,31 @@ from plotly.graph_objects import Figure
from plotly.io import to_json
from pydantic.fields import ModelField
from pynecone import constants, utils
from pynecone import constants
from pynecone.base import Base
from pynecone.utils import format, types
if TYPE_CHECKING:
from pynecone.state import State
# Set of unique variable names.
USED_VARIABLES = set()
def get_unique_variable_name() -> str:
"""Get a unique variable name.
Returns:
The unique variable name.
"""
name = "".join([random.choice(string.ascii_lowercase) for _ in range(8)])
if name not in USED_VARIABLES:
USED_VARIABLES.add(name)
return name
return get_unique_variable_name()
class Var(ABC):
"""An abstract var."""
@ -135,9 +155,9 @@ class Var(ABC):
Returns:
The wrapped var, i.e. {state.var}.
"""
out = self.full_name if self.is_local else utils.wrap(self.full_name, "{")
out = self.full_name if self.is_local else format.wrap(self.full_name, "{")
if self.is_string:
out = utils.format_string(out)
out = format.format_string(out)
return out
def __getitem__(self, i: Any) -> Var:
@ -154,8 +174,8 @@ class Var(ABC):
"""
# Indexing is only supported for lists, dicts, and dataframes.
if not (
utils._issubclass(self.type_, Union[List, Dict])
or utils.is_dataframe(self.type_)
types._issubclass(self.type_, Union[List, Dict])
or types.is_dataframe(self.type_)
):
if self.type_ == Any:
raise TypeError(
@ -173,9 +193,9 @@ class Var(ABC):
i = BaseVar(name=i.name, type_=i.type_, state=i.state, is_local=True)
# Handle list indexing.
if utils._issubclass(self.type_, List):
if types._issubclass(self.type_, List):
# List indices must be ints, slices, or vars.
if not isinstance(i, utils.get_args(Union[int, slice, Var])):
if not isinstance(i, types.get_args(Union[int, slice, Var])):
raise TypeError("Index must be an integer.")
# Handle slices first.
@ -192,10 +212,11 @@ class Var(ABC):
)
# Get the type of the indexed var.
if utils.is_generic_alias(self.type_):
type_ = utils.get_args(self.type_)[0]
else:
type_ = Any
type_ = (
types.get_args(self.type_)[0]
if types.is_generic_alias(self.type_)
else Any
)
# Use `at` to support negative indices.
return BaseVar(
@ -207,11 +228,10 @@ class Var(ABC):
# Dictionary / dataframe indexing.
# Get the type of the indexed var.
if isinstance(i, str):
i = utils.wrap(i, '"')
if utils.is_generic_alias(self.type_):
type_ = utils.get_args(self.type_)[1]
else:
type_ = Any
i = format.wrap(i, '"')
type_ = (
types.get_args(self.type_)[1] if types.is_generic_alias(self.type_) else Any
)
# Use normal indexing here.
return BaseVar(
@ -284,7 +304,7 @@ class Var(ABC):
props = (other, self) if flip else (self, other)
name = f"{props[0].full_name} {op} {props[1].full_name}"
if fn is None:
name = utils.wrap(name, "(")
name = format.wrap(name, "(")
if fn is not None:
name = f"{fn}({name})"
return BaseVar(
@ -337,7 +357,7 @@ class Var(ABC):
Raises:
TypeError: If the var is not a list.
"""
if not utils._issubclass(self.type_, List):
if not types._issubclass(self.type_, List):
raise TypeError(f"Cannot get length of non-list var {self}.")
return BaseVar(
name=f"{self.full_name}.length",
@ -607,7 +627,7 @@ class Var(ABC):
A var representing foreach operation.
"""
arg = BaseVar(
name=utils.get_unique_variable_name(),
name=get_unique_variable_name(),
type_=self.type_,
)
return BaseVar(
@ -685,13 +705,12 @@ class BaseVar(Var, Base):
Returns:
The default value of the var.
"""
if utils.is_generic_alias(self.type_):
type_ = self.type_.__origin__
else:
type_ = self.type_
type_ = (
self.type_.__origin__ if types.is_generic_alias(self.type_) else self.type_
)
if issubclass(type_, str):
return ""
if issubclass(type_, utils.get_args(Union[int, float])):
if issubclass(type_, types.get_args(Union[int, float])):
return 0
if issubclass(type_, bool):
return False

3
tests/compiler/test_compiler.py
View File

@ -3,6 +3,7 @@ from typing import Set
import pytest
from pynecone.compiler import utils
from pynecone.utils import imports
@pytest.mark.parametrize(
@ -42,7 +43,7 @@ def test_compile_import_statement(lib: str, fields: Set[str], output: str):
],
)
def test_compile_imports(
import_dict: utils.ImportDict, output: str, windows_platform: bool
import_dict: imports.ImportDict, output: str, windows_platform: bool
):
"""Test the compile_imports function.

4
tests/components/datadisplay/test_datatable.py
View File

@ -4,8 +4,8 @@ import pandas as pd
import pytest
import pynecone as pc
from pynecone import utils
from pynecone.components import data_table
from pynecone.utils import types
@pytest.mark.parametrize(
@ -33,7 +33,7 @@ def test_validate_data_table(data_table_state: pc.Var, expected):
"""
props = {"data": data_table_state.data}
if not utils.is_dataframe(data_table_state.data.type_):
if not types.is_dataframe(data_table_state.data.type_):
props["columns"] = data_table_state.columns
data_table_component = data_table(**props)

6
tests/components/media/test_icon.py
View File

@ -1,7 +1,7 @@
import pytest
from pynecone import utils
from pynecone.components.media.icon import ICON_LIST, Icon
from pynecone.utils import format
def test_no_tag_errors():
@ -27,7 +27,7 @@ def test_valid_icon(tag: str):
tag: The icon tag.
"""
icon = Icon.create(tag=tag)
assert icon.tag == utils.to_title_case(tag) + "Icon"
assert icon.tag == format.to_title_case(tag) + "Icon"
@pytest.mark.parametrize("tag", ["", " ", "invalid", 123])
@ -52,4 +52,4 @@ def test_tag_with_capital(tag: str):
tag: The icon tag.
"""
icon = Icon.create(tag=tag)
assert icon.tag == utils.to_title_case(tag) + "Icon"
assert icon.tag == format.to_title_case(tag) + "Icon"

7
tests/components/test_component.py
View File

@ -2,11 +2,12 @@ from typing import Dict, List, Type
import pytest
from pynecone.components.component import Component, CustomComponent, ImportDict
from pynecone.components.component import Component, CustomComponent
from pynecone.components.layout.box import Box
from pynecone.event import EVENT_ARG, EVENT_TRIGGERS, EventHandler
from pynecone.state import State
from pynecone.style import Style
from pynecone.utils import imports
from pynecone.var import Var
@ -39,7 +40,7 @@ def component1() -> Type[Component]:
# A test number prop.
number: Var[int]
def _get_imports(self) -> ImportDict:
def _get_imports(self) -> imports.ImportDict:
return {"react": {"Component"}}
def _get_custom_code(self) -> str:
@ -72,7 +73,7 @@ def component2() -> Type[Component]:
"on_close": EVENT_ARG,
}
def _get_imports(self) -> ImportDict:
def _get_imports(self) -> imports.ImportDict:
return {"react-redux": {"connect"}}
def _get_custom_code(self) -> str:

7
tests/test_event.py
View File

@ -1,7 +1,8 @@
import pytest
from pynecone import event, utils
from pynecone import event
from pynecone.event import Event, EventHandler, EventSpec
from pynecone.utils import format
from pynecone.var import Var
@ -57,8 +58,8 @@ def test_call_event_handler():
assert event_spec.handler == handler
assert event_spec.local_args == ()
assert event_spec.args == (
("arg1", utils.json_dumps(first)),
("arg2", utils.json_dumps(second)),
("arg1", format.json_dumps(first)),
("arg2", format.json_dumps(second)),
)
handler = EventHandler(fn=test_fn_with_args)

8
tests/test_state.py
View File

@ -3,11 +3,11 @@ from typing import Dict, List
import pytest
from plotly.graph_objects import Figure
from pynecone import utils
from pynecone.base import Base
from pynecone.constants import RouteVar
from pynecone.event import Event
from pynecone.state import State
from pynecone.utils import format
from pynecone.var import BaseVar, ComputedVar
@ -606,14 +606,14 @@ async def test_process_event_substate(test_state, child_state, grandchild_state)
def test_format_event_handler():
"""Test formatting an event handler."""
assert (
utils.format_event_handler(TestState.do_something) == "test_state.do_something" # type: ignore
format.format_event_handler(TestState.do_something) == "test_state.do_something" # type: ignore
)
assert (
utils.format_event_handler(ChildState.change_both) # type: ignore
format.format_event_handler(ChildState.change_both) # type: ignore
== "test_state.child_state.change_both"
)
assert (
utils.format_event_handler(GrandchildState.do_nothing) # type: ignore
format.format_event_handler(GrandchildState.do_nothing) # type: ignore
== "test_state.child_state.grandchild_state.do_nothing"
)

34
tests/test_utils.py
View File

@ -2,7 +2,7 @@ from typing import Any, List, Union
import pytest
from pynecone import utils
from pynecone.utils import build, format, imports, prerequisites, types
from pynecone.var import Var
@ -25,7 +25,7 @@ def test_to_snake_case(input: str, output: str):
input: The input string.
output: The expected output string.
"""
assert utils.to_snake_case(input) == output
assert format.to_snake_case(input) == output
@pytest.mark.parametrize(
@ -45,7 +45,7 @@ def test_to_camel_case(input: str, output: str):
input: The input string.
output: The expected output string.
"""
assert utils.to_camel_case(input) == output
assert format.to_camel_case(input) == output
@pytest.mark.parametrize(
@ -65,7 +65,7 @@ def test_to_title_case(input: str, output: str):
input: The input string.
output: The expected output string.
"""
assert utils.to_title_case(input) == output
assert format.to_title_case(input) == output
@pytest.mark.parametrize(
@ -86,7 +86,7 @@ def test_get_close_char(input: str, output: str):
input: The open character.
output: The expected close character.
"""
assert utils.get_close_char(input) == output
assert format.get_close_char(input) == output
@pytest.mark.parametrize(
@ -107,7 +107,7 @@ def test_is_wrapped(text: str, open: str, expected: bool):
open: The open character.
expected: Whether the text is wrapped.
"""
assert utils.is_wrapped(text, open) == expected
assert format.is_wrapped(text, open) == expected
@pytest.mark.parametrize(
@ -133,7 +133,7 @@ def test_wrap(text: str, open: str, expected: str, check_first: bool, num: int):
check_first: Whether to check if the text is already wrapped.
num: The number of times to wrap the text.
"""
assert utils.wrap(text, open, check_first=check_first, num=num) == expected
assert format.wrap(text, open, check_first=check_first, num=num) == expected
@pytest.mark.parametrize(
@ -155,7 +155,7 @@ def test_indent(text: str, indent_level: int, expected: str, windows_platform: b
expected: The expected output string.
windows_platform: Whether the system is windows.
"""
assert utils.indent(text, indent_level) == (
assert format.indent(text, indent_level) == (
expected.replace("\n", "\r\n") if windows_platform else expected
)
@ -176,14 +176,14 @@ def test_format_cond(condition: str, true_value: str, false_value: str, expected
false_value: The value to return if the condition is false.
expected: The expected output string.
"""
assert utils.format_cond(condition, true_value, false_value) == expected
assert format.format_cond(condition, true_value, false_value) == expected
def test_merge_imports():
"""Test that imports are merged correctly."""
d1 = {"react": {"Component"}}
d2 = {"react": {"Component"}, "react-dom": {"render"}}
d = utils.merge_imports(d1, d2)
d = imports.merge_imports(d1, d2)
assert set(d.keys()) == {"react", "react-dom"}
assert set(d["react"]) == {"Component"}
assert set(d["react-dom"]) == {"render"}
@ -207,7 +207,7 @@ def test_is_generic_alias(cls: type, expected: bool):
cls: The class to check.
expected: Whether the class is a GenericAlias.
"""
assert utils.is_generic_alias(cls) == expected
assert types.is_generic_alias(cls) == expected
@pytest.mark.parametrize(
@ -227,7 +227,7 @@ def test_format_route(route: str, expected: bool):
route: The route to format.
expected: The expected formatted route.
"""
assert utils.format_route(route) == expected
assert format.format_route(route) == expected
def test_setup_frontend(tmp_path, mocker):
@ -244,9 +244,11 @@ def test_setup_frontend(tmp_path, mocker):
assets.mkdir()
(assets / "favicon.ico").touch()
mocker.patch("pynecone.utils.install_frontend_packages")
assert str(web_folder) == prerequisites.create_web_directory(tmp_path)
utils.setup_frontend(tmp_path)
mocker.patch("pynecone.utils.prerequisites.install_frontend_packages")
build.setup_frontend(tmp_path)
assert web_folder.exists()
assert web_public_folder.exists()
assert (web_public_folder / "favicon.ico").exists()
@ -261,7 +263,7 @@ def test_setup_frontend(tmp_path, mocker):
],
)
def test_is_backend_variable(input, output):
assert utils.is_backend_variable(input) == output
assert types.is_backend_variable(input) == output
@pytest.mark.parametrize(
@ -282,4 +284,4 @@ def test_is_backend_variable(input, output):
],
)
def test_issubclass(cls: type, cls_check: type, expected: bool):
assert utils._issubclass(cls, cls_check) == expected
assert types._issubclass(cls, cls_check) == expected