lib

The lib module provides foundational serialization and utility classes. This is Layer 0 of the pyswark architecture - all other modules build on top of it.

Key Features

Type-preserving serialization: Serialize Pydantic models with embedded type information
Enhanced BaseModel: Standard base class for all pyswark models
Filesystem abstraction: Unified file system access via fsspec
Enum utilities: Extended enum functionality with alias support

Serialization (ser_des)

The ser_des module provides functions for serializing Pydantic models with embedded type information, enabling automatic deserialization without specifying the target type.

Serialization and Deserialization Utilities

This module provides type-preserving serialization for Pydantic models. The key feature is that serialized JSON embeds the model’s fully-qualified class name, enabling automatic deserialization without specifying the type.

Example

>>> from pyswark.lib.pydantic import ser_des
>>> from pydantic import BaseModel
>>>
>>> class Character(BaseModel):
...     name: str
...     role: str
>>>
>>> mulder = Character(name='Fox Mulder', role='FBI Agent')
>>> json_data = ser_des.toJson(mulder)
>>> restored = ser_des.fromJson(json_data)  # No type hint needed!
>>> assert type(restored) == Character

pyswark.lib.pydantic.ser_des.fromDict(data: dict) → Type[BaseModel]

Reconstruct a Pydantic model from a dictionary with type information.

Parameters:: data (dict) – Dictionary with model and contents keys.
Returns:: An instance of the original model class.
Return type:: BaseModel

See also

toDict: Create a typed dictionary from a model.

pyswark.lib.pydantic.ser_des.fromJson(loadable: str) → Type[BaseModel]

Deserialize JSON to its original Pydantic model type.

The JSON must have been created with toJson() (or have the same structure with model and contents keys).

Parameters:: loadable (str) – JSON string containing embedded type information.
Returns:: An instance of the original model class.
Return type:: BaseModel
Raises:: TypeError – If the model path cannot be resolved or is not a BaseModel subclass.

See also

toJson: Serialize a model with type information.
fromDict: Deserialize from a dictionary.

Example

>>> restored = fromJson(json_string)
>>> print(type(restored).__name__)
'Character'

pyswark.lib.pydantic.ser_des.toDict(model: Type[BaseModel]) → dict

Convert a Pydantic model to a dictionary with embedded type information.

Parameters:: model (BaseModel) – A Pydantic model instance.
Returns:: Dictionary with model (class path) and contents (model data) keys.
Return type:: dict

See also

fromDict: Reconstruct the model from the dictionary.

pyswark.lib.pydantic.ser_des.toJson(model: Type[BaseModel], **kw) → str

Serialize a Pydantic model to JSON with embedded type information.

The serialized output includes the model’s fully-qualified class name, enabling automatic deserialization via fromJson().

Parameters:

model (BaseModel) – A Pydantic model instance to serialize.
**kw – Additional keyword arguments passed to json.dumps() (e.g., indent=2 for pretty-printing).

Returns:

JSON string with structure: {"model": "module.Class", "contents": {...}}

Return type:

str

See also

fromJson: Deserialize JSON back to the original model type.
toDict: Convert to dict without JSON encoding.

Example

>>> json_str = toJson(my_model, indent=2)

Base Models

The base module provides enhanced Pydantic base classes.

Base Model Classes

This module provides enhanced Pydantic BaseModel classes with built-in serialization capabilities. All models in pyswark should extend from BaseModel defined here, not directly from pydantic.

Key Features

Automatic type-preserving serialization via toJson() / fromJson()
extra='forbid' by default to catch typos in field names
URI generation for model classes

class pyswark.lib.pydantic.base.BaseModel

Bases: ExtraForbidden

Enhanced Pydantic BaseModel with serialization capabilities.

This is the standard base class for all models in pyswark. It extends Pydantic’s BaseModel with type-preserving serialization methods that enable “code as data” patterns.

Example

>>> from pyswark.lib.pydantic import base
>>>
>>> class MyModel(base.BaseModel):
...     value: str
...     count: int
>>>
>>> obj = MyModel(value="hello", count=42)
>>> json_str = obj.toJson()  # Includes type information
>>> restored = base.BaseModel.fromJson(json_str)
>>> assert type(restored) == MyModel

static fromDict(*a, **kw)

Reconstruct a model from a dictionary with type information.

Parameters:

*a – Positional arguments passed to ser_des.fromDict().
**kw – Keyword arguments passed to ser_des.fromDict().

Returns:

The reconstructed model instance.

Return type:

BaseModel

static fromJson(*a, **kw)

Deserialize JSON to the original model type.

Parameters:

*a – Positional arguments passed to ser_des.fromJson().
**kw – Keyword arguments passed to ser_des.fromJson().

Returns:

The deserialized model instance.

Return type:

BaseModel

classmethod getUri()

Get the fully-qualified Python path for this model class.

Returns:: The module and class name (e.g., "mypackage.models.MyModel").
Return type:: str

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

toDict()

Convert this model to a dictionary with embedded type information.

Returns:: Dictionary with model and contents keys.
Return type:: dict

toJson(indent=2, **kw)

Serialize this model to JSON with embedded type information.

Parameters:

indent (int, optional) – JSON indentation level (default: 2).
**kw – Additional arguments passed to json.dumps().

Returns:

JSON string that can be deserialized with fromJson().

Return type:

str

write(uri, overwrite=False, indent=2, **kw): Write the model to a URI location.

class pyswark.lib.pydantic.base.ExtraAllowed(**extra_data: Any)

Bases: BaseModel

BaseModel variant that allows extra fields (flexible validation).

model_config: ClassVar[ConfigDict] = {'extra': 'allow'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

class pyswark.lib.pydantic.base.ExtraForbidden

Bases: BaseModel

BaseModel variant that forbids extra fields (strict validation).

model_config: ClassVar[ConfigDict] = {'extra': 'forbid'}: Configuration for the model, should be a dictionary conforming to [ConfigDict][pydantic.config.ConfigDict].

Enum Utilities

The enum module provides enhanced enum classes with utility methods.

Enum Utilities

This module provides enhanced enum classes with additional utility methods for introspection, code generation, and dynamic enum creation.

Classes

Enum: Enhanced enum with utility methods (asDict, asPythonCode, createDynamically).
Mixin: Mixin class providing utility methods for enums.

class pyswark.lib.enum.Enum(value, names=<not given>, *values, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Mixin, Enum

Enhanced enum with utility methods.

Extends Python’s built-in Enum with additional methods for introspection and code generation.

Example

>>> from pyswark.lib.enum import Enum
>>>
>>> class Status(Enum):
...     ACTIVE = 1
...     INACTIVE = 0
>>>
>>> Status.asDict()
{'ACTIVE': 1, 'INACTIVE': 0}

class pyswark.lib.enum.Mixin

Bases: object

Mixin providing utility methods for enum classes.

asDict(): Return a dict mapping member names to values.

asPythonCode(name): Generate Python code to recreate the enum.

createDynamically(enumMembers, enumName): Create an enum dynamically from a dict.

classmethod asDict()

Return a dict mapping member names to values.

Returns:: Dictionary with member names as keys and values as values.
Return type:: dict

Example

>>> class Status(Enum):
...     ACTIVE = 1
...     INACTIVE = 0
>>> Status.asDict()
{'ACTIVE': 1, 'INACTIVE': 0}

classmethod asPythonCode(name='')

Generate Python code to recreate the enum.

Parameters:: name (str, optional) – Name for the generated class. Defaults to My{ClassName}.
Returns:: Python source code string.
Return type:: str

classmethod createDynamically(enumMembers, enumName='')

Create an enum dynamically from a dict of members.

Parameters:

enumMembers (dict) – Dictionary mapping member names to values.
enumName (str, optional) – Name for the new enum class.

Returns:

A new enum class with the specified members.

Return type:

Enum

Example

>>> MyEnum = Enum.createDynamically({'A': 1, 'B': 2}, 'MyEnum')
>>> MyEnum.A.value
1

AliasEnum

The aenum module provides enum classes with alias support, allowing multiple names to reference the same member.

Advanced Enum with Aliases

This module provides an enhanced enum class that supports aliases, allowing multiple names to refer to the same enum member.

Classes

AliasEnum: Enum class supporting aliases for members.
Alias: Helper class to define aliases for enum members.

Example

>>> from pyswark.lib.aenum import AliasEnum, Alias
>>>
>>> class Protocol(AliasEnum):
...     HTTP = 80, Alias('http', 'web')
...     HTTPS = 443, Alias('https', 'secure')
...     SSH = 22
>>>
>>> Protocol.get('http')  # Returns Protocol.HTTP
>>> Protocol.get('secure')  # Returns Protocol.HTTPS

class pyswark.lib.aenum.Alias(alias, *aliases)

Bases: object

Container for alias values.

Holds a set of aliases that can be used with AliasEnum to allow multiple names to reference the same enum member.

Parameters:

alias (str or list or tuple or set) – Primary alias or collection of aliases.
*aliases (str) – Additional aliases.

valueset

The set of all aliases.

Type:: set

Example

>>> Alias('a', 'b').valueset
{'a', 'b'}
>>> Alias(['a', 'b']).valueset
{'a', 'b'}

property valueset

class pyswark.lib.aenum.AliasEnum(value=<no_arg>, names=None, module=None, qualname=None, type=None, start=1, boundary=None)

Bases: Mixin, Enum

Enum class supporting aliases for members.

Allows multiple names to reference the same enum member, enabling flexible lookups by alias, name, or member.

get(aliasOrMember): Get member by alias, name, or member instance.

tryGet(aliasOrMember, default=None): Like get() but returns default instead of raising.

toMapping(attr='value'): Create a dict mapping all aliases to member values.

Example

>>> class Protocol(AliasEnum):
...     HTTP = 80, Alias('http', 'web')
...     HTTPS = 443, Alias('https', 'secure')
...     SSH = 22
>>>
>>> Protocol.get('http').value
80
>>> Protocol.get('secure').value
443
>>> Protocol.HTTP.aliases
{'HTTP', 'http', 'web'}

property aliases

classmethod get(aliasOrMember)

Get an enum member by alias, name, or member instance.

Parameters:: aliasOrMember (str or AliasEnum) – The alias, name, or member to look up.
Returns:: The matching enum member.
Return type:: AliasEnum
Raises:: AliasEnumError – If no member matches the alias.

property name

classmethod toMapping(attr='value')

Create a mapping from all aliases to member attribute values.

Parameters:: attr (str, default='value') – The attribute to use as the mapping value.
Returns:: Dictionary mapping all aliases (including names) to values.
Return type:: dict

classmethod tryGet(aliasOrMember, default=None)

Get member by alias, returning default if not found.

Parameters:

aliasOrMember (str or AliasEnum) – The alias, name, or member to look up.
default (Any, optional) – Value to return if not found.

Returns:

The matching member or the default value.

Return type:

AliasEnum or default

property value

exception pyswark.lib.aenum.AliasEnumError

Bases: Exception

Exception raised when an alias lookup fails.

Filesystem (fsspec)

The fsspec submodule provides unified filesystem access with custom protocol support.

Filesystem Abstraction

This module provides a unified filesystem interface built on fsspec, with additional protocol support (e.g., python:// for Python objects).

Functions

open(uri): Open a file from any supported URI.
filesystem(protocol): Get a filesystem instance for a protocol.

pyswark.lib.fsspec.fsspec.filesystem(protocol, **kwargs)

Get a filesystem instance for a protocol.

Parameters:

protocol (str) – The protocol name (e.g., ‘file’, ‘python’, ‘s3’).
**kwargs – Additional arguments passed to the filesystem.

Returns:

A filesystem instance.

Return type:

AbstractFileSystem

pyswark.lib.fsspec.fsspec.open(uri, *args, **kwargs)

Open a file from any supported URI.

Wraps fsspec.open with additional protocol support.

Parameters:

uri (str) – The URI to open.
*args – Additional positional arguments passed to fsspec.
**kwargs – Additional keyword arguments passed to fsspec.

Returns:

A file-like object.

Return type:

OpenFile

Filesystem Implementations

This module provides custom fsspec filesystem implementations, including the PythonFileSystem for accessing Python objects by path.

class pyswark.lib.fsspec.implementations.PythonFile(path, *args, data=None)

Bases: object

File-like wrapper for Python objects.

Provides a context manager interface for accessing Python objects located by import path.

Parameters:

path (str) – The Python import path (e.g., ‘module.submodule.ClassName’).
data (Any, optional) – Pre-loaded data (if already located).

close()

locate(reloadmodule=False, **kw)

Locate and return the Python object.

Parameters:

reloadmodule (bool, default=False) – If True, reload the module before locating.
**kw – Additional arguments passed to pydoc.locate.

Returns:

The located Python object.

Return type:

Any

class pyswark.lib.fsspec.implementations.PythonFileSystem(*args, **kwargs)

Bases: AbstractFileSystem

Filesystem for accessing Python objects by import path.

Allows treating Python objects (classes, functions, modules) as files that can be “opened” and “read” via their import paths.

Example

>>> fs = PythonFileSystem()
>>> fs.exists('os.path.join')
True
>>> obj = fs.open('json.loads').locate()

exists(path, returnDataToo=False, **kw): Is there a file at the given path

open(path, *args, **kw)

Return a file-like object from the filesystem

The resultant instance must function correctly in a context with block.

Parameters:

path (str) – Target file
mode (str like 'rb', 'w') – See builtin open() Mode “x” (exclusive write) may be implemented by the backend. Even if it is, whether it is checked up front or on commit, and whether it is atomic is implementation-dependent.
block_size (int) – Some indication of buffering - this is a value in bytes
cache_options (dict, optional) – Extra arguments to pass through to the cache.
compression (string or None) – If given, open file using compression codec. Can either be a compression name (a key in fsspec.compression.compr) or “infer” to guess the compression from the filename suffix.
encoding (passed on to TextIOWrapper for text mode) –
errors (passed on to TextIOWrapper for text mode) –
newline (passed on to TextIOWrapper for text mode) –

Usage Examples

Type-Preserving Serialization

from pydantic import BaseModel
from pyswark.lib.pydantic import ser_des

class Character(BaseModel):
    name: str
    role: str

# Create an instance
mulder = Character(name='Fox Mulder', role='FBI Agent')

# Serialize with type information
json_data = ser_des.toJson(mulder, indent=2)
# Output:
# {
#   "model": "__main__.Character",
#   "contents": {
#     "name": "Fox Mulder",
#     "role": "FBI Agent"
#   }
# }

# Deserialize - no type hint needed!
restored = ser_des.fromJson(json_data)
assert type(restored) == Character

Creating Custom Models

from pyswark.lib.pydantic import base

class MyModel(base.BaseModel):
    value: str
    count: int

obj = MyModel(value="hello", count=42)

# Built-in serialization methods
json_str = obj.toJson()
restored = base.BaseModel.fromJson(json_str)

Using Enum Utilities

from pyswark.lib.enum import Enum

class Status(Enum):
    ACTIVE = 1
    INACTIVE = 0
    PENDING = 2

# Get dict of all members
Status.asDict()
# {'ACTIVE': 1, 'INACTIVE': 0, 'PENDING': 2}

# Generate Python code
print(Status.asPythonCode('MyStatus'))

# Create enum dynamically
DynamicStatus = Enum.createDynamically(
    {'ON': True, 'OFF': False},
    'DynamicStatus'
)

Using AliasEnum

from pyswark.lib.aenum import AliasEnum, Alias

class Protocol(AliasEnum):
    HTTP = 80, Alias('http', 'web')
    HTTPS = 443, Alias('https', 'secure')
    SSH = 22

# Access by name
Protocol.HTTP.value  # 80

# Access by alias
Protocol.get('http').value   # 80
Protocol.get('web').value    # 80
Protocol.get('secure').value # 443

# View aliases
Protocol.HTTP.aliases  # {'HTTP', 'http', 'web'}

# Create mapping from all aliases to values
Protocol.toMapping()
# {'HTTP': 80, 'http': 80, 'web': 80, 'HTTPS': 443, ...}

# Safe lookup with default
Protocol.tryGet('unknown', default=None)  # None

Using Filesystem Abstraction

from pyswark.lib.fsspec import fsspec

# Get a filesystem for a protocol
fs = fsspec.filesystem('python')

# Check if a Python object exists
fs.exists('os.path.join')  # True

# Open and locate a Python object
with fs.open('json.loads') as f:
    loads_func = f.locate()