#sqlalchemy-pydantic-mapper
`sqlalchemy-pydantic-mapper` simplifies converting SQLAlchemy instances (subclasses of `sqlalchemy.orm.DeclarativeBase`) into Pydantic models (`pydantic.BaseModel`).
#### It supports:
* registering custom synchronous and asynchronous mappers;
* registration via decorator or by passing `func=` directly;
* automatic mapping via Pydantic if the model has `model_config = ConfigDict(from_attributes=True)`;
* `map(...)` — an async method returning the target Pydantic model instance (must `await` it);
* `map_each(...)` — an async method returning the sequence of target Pydantic model instances, calls target func for every ORM model (must `await` it);
* `map_bulk(...)` — an async method returning a list of mapped Pydantic model instances from a sequence of SQLAlchemy objects(must `await` it).
---
## Usage Examples (Full Code Snippets)
1. Simple registration via `func=` and checking `_mappers`:
```python
from sqlalchemy_pydantic_mapper import ObjectMapper
def mapper(db: UserDB) -> UserSchema:
return UserSchema(id=db.id, name=db.name)
ObjectMapper.register(UserDB, UserSchema, func=mapper)
assert ObjectMapper._mappers_single[UserDB][UserSchema] is mapper
```
2. Registration via decorator:
```python
@ObjectMapper.register(UserDB, UserSchema)
def mapper2(db: UserDB) -> UserSchema:
return UserSchema(id=db.id, name=db.name)
```
3. Async mapper (registration + usage):
```python
@ObjectMapper.register(UserDB, UserSchema)
async def async_mapper(db: UserDB) -> UserSchema:
# e.g., async call or await something
return UserSchema(id=db.id, name=db.name.upper())
import asyncio
async def main():
user = UserDB()
user.id = 1
user.name = "alice"
res = await ObjectMapper.map(user, UserSchema)
print(res) # UserSchema(id=1, name='ALICE')
asyncio.run(main())
```
4. Auto-mapping via Pydantic (`from_attributes=True`):
```python
class UserSchema(BaseModel):
model_config = ConfigDict(from_attributes=True)
id: int
name: str
```
If no custom mapper is registered for the `from_class -> to_class` pair, `ObjectMapper.map(instance, UserSchema)` automatically calls:
```python
UserSchema.model_validate(instance, from_attributes=True)
```
5. Example in a test (synthetic):
```python
async def test_auto_mapping():
stud = StudentDB()
stud.id = 2
stud.name = "Bob"
result = await ObjectMapper.map(stud, UserSchema)
assert result.id == 2
assert result.name == "Bob"
```
6. Example for manual mapping (different names / logic):
```python
def stud_to_studschema(db: StudentDB) -> StudSchema:
return StudSchema(id=db.id, name=db.name)
ObjectMapper.register(StudentDB, StudSchema, func=stud_to_studschema)
```
---
7. Mapping multiple objects with `map_many`
```python
users = [UserDB(id=1, name="Alice"), UserDB(id=2, name="Bob")]
# Synchronous mapper
def mapper(db: UserDB) -> UserSchema:
return UserSchema(id=db.id, name=db.name)
ObjectMapper.register(UserDB, UserSchema, func=mapper)
import asyncio
async def main():
results = await ObjectMapper.map_each(users, UserSchema)
print(results)
# [UserSchema(id=1, name='Alice'), UserSchema(id=2, name='Bob')]
asyncio.run(main())
```
---
8. Passing additional arguments to a mapper function
```python
def mapper_with_prefix(db: UserDB, prefix: str) -> UserSchema:
return UserSchema(id=db.id, name=f"{prefix}{db.name}")
# Register mapper with a closure to pass extra arguments
ObjectMapper.register(UserDB, UserSchema, func=lambda db: mapper_with_prefix(db, prefix="Mr. "))
user = UserDB(id=1, name="Alice")
import asyncio
async def main():
result = await ObjectMapper.map(user, UserSchema)
print(result) # UserSchema(id=1, name='Mr. Alice')
asyncio.run(main())
```
---
9. Re-registering a mapper (overwriting)
```python
# Original mapper
ObjectMapper.register(UserDB, UserSchema, func=lambda db: UserSchema(id=db.id, name=db.name))
# Re-register with a new logic
ObjectMapper.register(UserDB, UserSchema,
func=lambda db: UserSchema(id=db.id, name=db.name.upper()),
override_existing=True)
user = UserDB(id=2, name="Bob")
import asyncio
async def main():
result = await ObjectMapper.map(user, UserSchema)
print(result) # UserSchema(id=2, name='BOB')
asyncio.run(main())
```
---
10. Async mapper with `map_each`
```python
async def async_mapper(db: UserDB) -> UserSchema:
import asyncio
await asyncio.sleep(0.01)
return UserSchema(id=db.id, name=db.name.upper())
ObjectMapper.register(UserDB, UserSchema, func=async_mapper)
users = [UserDB(id=1, name="Alice"), UserDB(id=2, name="Bob")]
async def main():
results = await ObjectMapper.map_each(users, UserSchema)
print(results)
# [UserSchema(id=1, name='ALICE'), UserSchema(id=2, name='BOB')]
asyncio.run(main())
```
11. Async mapper with `map_bulk` - True bulk operation under all the sequence at once
```python
async def async_mapper(dbs: Sequence[UserDB]) -> Sequence[UserSchema]:
import asyncio
await asyncio.sleep(0.01)
return [UserSchema(id=db.id, name=db.name.upper()) for db in dbs]
ObjectMapper.register_bulk(UserDB, UserSchema, func=async_mapper)
users = [UserDB(id=1, name="Alice"), UserDB(id=2, name="Bob")]
async def main():
results = await ObjectMapper.map_bulk(users, UserSchema)
print(results)
# [UserSchema(id=1, name='ALICE'), UserSchema(id=2, name='BOB')]
asyncio.run(main())
```
## Errors and Behavior on Incorrect Usage
* `TypeError` if `from_` does not inherit `DeclarativeBase`:
```python
class NotABase: pass
ObjectMapper.register(NotABase, UserSchema) # -> TypeError
```
* `TypeError` if `to_` does not inherit `BaseModel`:
```python
class NotABaseModel: pass
ObjectMapper.register(UserDB, NotABaseModel) # -> TypeError
```
* `ValueError` if `func` is missing and `to_` does not have `model_config = ConfigDict(from_attributes=True)`:
```python
class BadSchema(BaseModel):
id: int
name: str
ObjectMapper.register(UserDB, BadSchema) # -> ValueError
```
---
Raw data
{
"_id": null,
"home_page": null,
"name": "sqlalchemy-pydantic-mapper",
"maintainer": null,
"docs_url": null,
"requires_python": ">=3.13",
"maintainer_email": null,
"keywords": "sqlalchemy, pydantic, mapper, async",
"author": "ItzSkyReed",
"author_email": null,
"download_url": "https://files.pythonhosted.org/packages/4c/8d/b7804d194f4eed7ee3fa325d15d29a1f45231e4fbbddf83a77393667be4d/sqlalchemy_pydantic_mapper-0.3.0.tar.gz",
"platform": null,
"description": "#sqlalchemy-pydantic-mapper\n\n`sqlalchemy-pydantic-mapper` simplifies converting SQLAlchemy instances (subclasses of `sqlalchemy.orm.DeclarativeBase`) into Pydantic models (`pydantic.BaseModel`).\n\n#### It supports:\n\n* registering custom synchronous and asynchronous mappers;\n* registration via decorator or by passing `func=` directly;\n* automatic mapping via Pydantic if the model has `model_config = ConfigDict(from_attributes=True)`;\n* `map(...)` \u2014 an async method returning the target Pydantic model instance (must `await` it);\n* `map_each(...)` \u2014 an async method returning the sequence of target Pydantic model instances, calls target func for every ORM model (must `await` it);\n* `map_bulk(...)` \u2014 an async method returning a list of mapped Pydantic model instances from a sequence of SQLAlchemy objects(must `await` it).\n\n---\n## Usage Examples (Full Code Snippets)\n\n1. Simple registration via `func=` and checking `_mappers`:\n\n```python\nfrom sqlalchemy_pydantic_mapper import ObjectMapper\n\n\ndef mapper(db: UserDB) -> UserSchema:\n return UserSchema(id=db.id, name=db.name)\n\n\nObjectMapper.register(UserDB, UserSchema, func=mapper)\nassert ObjectMapper._mappers_single[UserDB][UserSchema] is mapper\n```\n\n2. Registration via decorator:\n\n```python\n@ObjectMapper.register(UserDB, UserSchema)\ndef mapper2(db: UserDB) -> UserSchema:\n return UserSchema(id=db.id, name=db.name)\n```\n\n3. Async mapper (registration + usage):\n\n```python\n@ObjectMapper.register(UserDB, UserSchema)\nasync def async_mapper(db: UserDB) -> UserSchema:\n # e.g., async call or await something\n return UserSchema(id=db.id, name=db.name.upper())\n\nimport asyncio\n\nasync def main():\n user = UserDB()\n user.id = 1\n user.name = \"alice\"\n res = await ObjectMapper.map(user, UserSchema)\n print(res) # UserSchema(id=1, name='ALICE')\n\nasyncio.run(main())\n```\n\n4. Auto-mapping via Pydantic (`from_attributes=True`):\n\n```python\nclass UserSchema(BaseModel):\n model_config = ConfigDict(from_attributes=True)\n id: int\n name: str\n```\n\nIf no custom mapper is registered for the `from_class -> to_class` pair, `ObjectMapper.map(instance, UserSchema)` automatically calls:\n\n```python\nUserSchema.model_validate(instance, from_attributes=True)\n```\n\n5. Example in a test (synthetic):\n\n```python\nasync def test_auto_mapping():\n stud = StudentDB()\n stud.id = 2\n stud.name = \"Bob\"\n\n result = await ObjectMapper.map(stud, UserSchema)\n assert result.id == 2\n assert result.name == \"Bob\"\n```\n\n6. Example for manual mapping (different names / logic):\n\n```python\ndef stud_to_studschema(db: StudentDB) -> StudSchema:\n return StudSchema(id=db.id, name=db.name)\n\nObjectMapper.register(StudentDB, StudSchema, func=stud_to_studschema)\n```\n---\n\n\n7. Mapping multiple objects with `map_many`\n\n```python\nusers = [UserDB(id=1, name=\"Alice\"), UserDB(id=2, name=\"Bob\")]\n\n\n# Synchronous mapper\ndef mapper(db: UserDB) -> UserSchema:\n return UserSchema(id=db.id, name=db.name)\n\n\nObjectMapper.register(UserDB, UserSchema, func=mapper)\n\nimport asyncio\n\n\nasync def main():\n results = await ObjectMapper.map_each(users, UserSchema)\n print(results)\n # [UserSchema(id=1, name='Alice'), UserSchema(id=2, name='Bob')]\n\n\nasyncio.run(main())\n```\n\n---\n\n8. Passing additional arguments to a mapper function\n\n```python\ndef mapper_with_prefix(db: UserDB, prefix: str) -> UserSchema:\n return UserSchema(id=db.id, name=f\"{prefix}{db.name}\")\n\n# Register mapper with a closure to pass extra arguments\nObjectMapper.register(UserDB, UserSchema, func=lambda db: mapper_with_prefix(db, prefix=\"Mr. \"))\n\nuser = UserDB(id=1, name=\"Alice\")\nimport asyncio\n\nasync def main():\n result = await ObjectMapper.map(user, UserSchema)\n print(result) # UserSchema(id=1, name='Mr. Alice')\n\nasyncio.run(main())\n```\n\n---\n\n9. Re-registering a mapper (overwriting)\n\n```python\n# Original mapper\nObjectMapper.register(UserDB, UserSchema, func=lambda db: UserSchema(id=db.id, name=db.name))\n\n# Re-register with a new logic\nObjectMapper.register(UserDB, UserSchema, \n func=lambda db: UserSchema(id=db.id, name=db.name.upper()),\n override_existing=True)\n\nuser = UserDB(id=2, name=\"Bob\")\n\nimport asyncio\nasync def main():\n result = await ObjectMapper.map(user, UserSchema)\n print(result) # UserSchema(id=2, name='BOB')\n\nasyncio.run(main())\n```\n\n---\n\n10. Async mapper with `map_each`\n\n```python\nasync def async_mapper(db: UserDB) -> UserSchema:\n import asyncio\n await asyncio.sleep(0.01)\n return UserSchema(id=db.id, name=db.name.upper())\n\n\nObjectMapper.register(UserDB, UserSchema, func=async_mapper)\n\nusers = [UserDB(id=1, name=\"Alice\"), UserDB(id=2, name=\"Bob\")]\n\n\nasync def main():\n results = await ObjectMapper.map_each(users, UserSchema)\n print(results)\n # [UserSchema(id=1, name='ALICE'), UserSchema(id=2, name='BOB')]\n\n\nasyncio.run(main())\n```\n\n11. Async mapper with `map_bulk` - True bulk operation under all the sequence at once\n\n```python\nasync def async_mapper(dbs: Sequence[UserDB]) -> Sequence[UserSchema]:\n import asyncio\n await asyncio.sleep(0.01)\n return [UserSchema(id=db.id, name=db.name.upper()) for db in dbs]\n\n\nObjectMapper.register_bulk(UserDB, UserSchema, func=async_mapper)\n\nusers = [UserDB(id=1, name=\"Alice\"), UserDB(id=2, name=\"Bob\")]\n\n\nasync def main():\n results = await ObjectMapper.map_bulk(users, UserSchema)\n print(results)\n # [UserSchema(id=1, name='ALICE'), UserSchema(id=2, name='BOB')]\n\n\nasyncio.run(main())\n```\n\n\n## Errors and Behavior on Incorrect Usage\n\n* `TypeError` if `from_` does not inherit `DeclarativeBase`:\n\n```python\nclass NotABase: pass\nObjectMapper.register(NotABase, UserSchema) # -> TypeError\n```\n\n* `TypeError` if `to_` does not inherit `BaseModel`:\n\n```python\nclass NotABaseModel: pass\nObjectMapper.register(UserDB, NotABaseModel) # -> TypeError\n```\n\n* `ValueError` if `func` is missing and `to_` does not have `model_config = ConfigDict(from_attributes=True)`:\n\n```python\nclass BadSchema(BaseModel):\n id: int\n name: str\n\nObjectMapper.register(UserDB, BadSchema) # -> ValueError\n```\n---\n",
"bugtrack_url": null,
"license": "MIT",
"summary": "sqlalchemy-pydantic-mapper library for mapping SQLAlchemy models to Pydantic",
"version": "0.3.0",
"project_urls": {
"Homepage": "https://github.com/ItzSkyReed/sqlalchemy-pydantic-mapper"
},
"split_keywords": [
"sqlalchemy",
" pydantic",
" mapper",
" async"
],
"urls": [
{
"comment_text": null,
"digests": {
"blake2b_256": "3250c6aa16d60156693868675402d1370b26eee39d3e541e122c22da6009d8a8",
"md5": "8a341343a092857ff4cbd3bdef5ed78c",
"sha256": "1cb9ec7b7a2403811f78a17b184a2d9d91c6d0f4fa9416555b369d66a54ae60c"
},
"downloads": -1,
"filename": "sqlalchemy_pydantic_mapper-0.3.0-py3-none-any.whl",
"has_sig": false,
"md5_digest": "8a341343a092857ff4cbd3bdef5ed78c",
"packagetype": "bdist_wheel",
"python_version": "py3",
"requires_python": ">=3.13",
"size": 7975,
"upload_time": "2025-08-30T17:32:05",
"upload_time_iso_8601": "2025-08-30T17:32:05.102615Z",
"url": "https://files.pythonhosted.org/packages/32/50/c6aa16d60156693868675402d1370b26eee39d3e541e122c22da6009d8a8/sqlalchemy_pydantic_mapper-0.3.0-py3-none-any.whl",
"yanked": false,
"yanked_reason": null
},
{
"comment_text": null,
"digests": {
"blake2b_256": "4c8db7804d194f4eed7ee3fa325d15d29a1f45231e4fbbddf83a77393667be4d",
"md5": "99199f6ac636109a7379f09f4c09ef6c",
"sha256": "1090f78fd58f3094896774782b7d64ed733074d7e74fdd5fbb0e0910ebb5a1c2"
},
"downloads": -1,
"filename": "sqlalchemy_pydantic_mapper-0.3.0.tar.gz",
"has_sig": false,
"md5_digest": "99199f6ac636109a7379f09f4c09ef6c",
"packagetype": "sdist",
"python_version": "source",
"requires_python": ">=3.13",
"size": 10981,
"upload_time": "2025-08-30T17:32:06",
"upload_time_iso_8601": "2025-08-30T17:32:06.433167Z",
"url": "https://files.pythonhosted.org/packages/4c/8d/b7804d194f4eed7ee3fa325d15d29a1f45231e4fbbddf83a77393667be4d/sqlalchemy_pydantic_mapper-0.3.0.tar.gz",
"yanked": false,
"yanked_reason": null
}
],
"upload_time": "2025-08-30 17:32:06",
"github": true,
"gitlab": false,
"bitbucket": false,
"codeberg": false,
"github_user": "ItzSkyReed",
"github_project": "sqlalchemy-pydantic-mapper",
"travis_ci": false,
"coveralls": false,
"github_actions": true,
"lcname": "sqlalchemy-pydantic-mapper"
}
JSON
