pi3c
/
test_api
1
0
Fork 0
Code Issues Pull Requests Packages Projects Releases Wiki Activity

just sync

main
Сергей Ванюшкин 2024-03-17 09:46:58 +00:00
parent 067a064cf1
commit d2bc188fac
5 changed files with 54 additions and 9 deletions
Show Stats Download Patch File Download Diff File Expand all files Collapse all files

37
poetry.lock generated
View File

@ -425,6 +425,23 @@ files = [
[package.dependencies]
setuptools = "*"
[[package]]
name = "passlib"
version = "1.7.4"
description = "comprehensive password hashing framework supporting over 30 schemes"
optional = false
python-versions = "*"
files = [
{file = "passlib-1.7.4-py2.py3-none-any.whl", hash = "sha256:aa6bca462b8d8bda89c70b382f0c298a20b5560af6cbfa2dce410c0a2fb669f1"},
{file = "passlib-1.7.4.tar.gz", hash = "sha256:defd50f72b65c5402ab2c573830a6978e5f202ad0d984793c8dde2c4152ebe04"},
]
[package.extras]
argon2 = ["argon2-cffi (>=18.2.0)"]
bcrypt = ["bcrypt (>=3.1.0)"]
build-docs = ["cloud-sptheme (>=1.10.1)", "sphinx (>=1.6)", "sphinxcontrib-fulltoc (>=1.2.0)"]
totp = ["cryptography"]
[[package]]
name = "platformdirs"
version = "4.2.0"
@ -668,6 +685,23 @@ python-dotenv = ">=0.21.0"
toml = ["tomli (>=2.0.1)"]
yaml = ["pyyaml (>=6.0.1)"]
[[package]]
name = "pyjwt"
version = "2.8.0"
description = "JSON Web Token implementation in Python"
optional = false
python-versions = ">=3.7"
files = [
{file = "PyJWT-2.8.0-py3-none-any.whl", hash = "sha256:59127c392cc44c2da5bb3192169a91f429924e17aff6534d70fdc02ab3e04320"},
{file = "PyJWT-2.8.0.tar.gz", hash = "sha256:57e28d156e3d5c10088e0c68abb90bfac3df82b40a71bd0daa20c65ccd5c23de"},
]
[package.extras]
crypto = ["cryptography (>=3.4.0)"]
dev = ["coverage[toml] (==5.0.4)", "cryptography (>=3.4.0)", "pre-commit", "pytest (>=6.0.0,<7.0.0)", "sphinx (>=4.5.0,<5.0.0)", "sphinx-rtd-theme", "zope.interface"]
docs = ["sphinx (>=4.5.0,<5.0.0)", "sphinx-rtd-theme", "zope.interface"]
tests = ["coverage[toml] (==5.0.4)", "pytest (>=6.0.0,<7.0.0)"]
[[package]]
name = "python-dotenv"
version = "1.0.1"
@ -707,6 +741,7 @@ files = [
{file = "PyYAML-6.0.1-cp311-cp311-win_amd64.whl", hash = "sha256:bf07ee2fef7014951eeb99f56f39c9bb4af143d8aa3c21b1677805985307da34"},
{file = "PyYAML-6.0.1-cp312-cp312-macosx_10_9_x86_64.whl", hash = "sha256:855fb52b0dc35af121542a76b9a84f8d1cd886ea97c84703eaa6d88e37a2ad28"},
{file = "PyYAML-6.0.1-cp312-cp312-macosx_11_0_arm64.whl", hash = "sha256:40df9b996c2b73138957fe23a16a4f0ba614f4c0efce1e9406a184b6d07fa3a9"},
{file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_aarch64.manylinux2014_aarch64.whl", hash = "sha256:a08c6f0fe150303c1c6b71ebcd7213c2858041a7e01975da3a99aed1e7a378ef"},
{file = "PyYAML-6.0.1-cp312-cp312-manylinux_2_17_x86_64.manylinux2014_x86_64.whl", hash = "sha256:6c22bec3fbe2524cde73d7ada88f6566758a8f7227bfbf93a408a9d86bcc12a0"},
{file = "PyYAML-6.0.1-cp312-cp312-musllinux_1_1_x86_64.whl", hash = "sha256:8d4e9c88387b0f5c7d5f281e55304de64cf7f9c0021a3525bd3b1c542da3b0e4"},
{file = "PyYAML-6.0.1-cp312-cp312-win32.whl", hash = "sha256:d483d2cdf104e7c9fa60c544d92981f12ad66a457afae824d146093b8c294c54"},
@ -925,4 +960,4 @@ test = ["covdefaults (>=2.3)", "coverage (>=7.2.7)", "coverage-enable-subprocess
[metadata]
lock-version = "2.0"
python-versions = "^3.10"
content-hash = "9b28e04ae14dfa77f629d230f28168647f584a9c28b97367b7c238374e1b1049"
content-hash = "59ca9e4d75435b8ae029ef26e5d01c7e215cba33e44bdd9967cbbc540bd01226"

2
pyproject.toml
View File

@ -9,6 +9,8 @@ readme = "README.md"
python = "^3.10"
sqlalchemy = "^2.0.28"
asyncpg = "^0.29.0"
passlib = "^1.7.4"
pyjwt = "^2.8.0"
[tool.poetry.scripts]
api = "test_api.run:run_server"

6
test_api/di.py
View File

@ -1,9 +1,13 @@
from sqlalchemy.ext.asyncio import AsyncSession, async_sessionmaker, create_async_engine
from fastapi.security import OAuth2PasswordBearer
from sqlalchemy.ext.asyncio import (AsyncSession, async_sessionmaker,
create_async_engine)
from .config import get_settings
from .services.user import UserService
from .uow.uow_base import UnitOfWork
Oauth2_scheme = OAuth2PasswordBearer(tokenUrl="/token")
async_engine = create_async_engine(
url=get_settings().db.get_db_url,
echo=False,

11
test_api/repositories/user.py
View File

@ -1,3 +1,4 @@
from passlib.context import CryptContext
from sqlalchemy import delete, insert, select, update
from sqlalchemy.ext.asyncio.session import AsyncSession
@ -9,10 +10,13 @@ class UserRepository:
def __init__(self, session: AsyncSession):
self.session = session
self.crypto_context = CryptContext(schemes="bcrypt")
async def add_one(
self,
data: UserWriteDTO,
) -> UserReadDTO:
data.hashed_password = self.crypto_context.hash(data.hashed_password)
stmt = insert(UserModel).values(**data.model_dump()).returning(UserModel)
res = await self.session.execute(stmt)
return UserReadDTO.model_validate(res.scalar_one())
@ -34,7 +38,12 @@ class UserRepository:
return None
async def patch_one(self, filter: dict, data: UserWriteDTO) -> UserReadDTO | None:
stmt = update(UserModel).where(UserModel.id == filter["id"]).values(**data.model_dump()).returning(UserModel)
stmt = (
update(UserModel)
.where(UserModel.id == filter["id"])
.values(**data.model_dump())
.returning(UserModel)
)
res = await self.session.execute(stmt)
res = res.scalar_one_or_none()

7
test_api/services/user.py
View File

@ -23,7 +23,6 @@ class UserService:
return res
async def add_one(self, data: UserWriteDTO):
res = None
async with self.uow:
try:
res = await self.uow.users.add_one(data)
@ -48,7 +47,6 @@ class UserService:
return res
async def patch_one(self, id: UUID, data: UserWriteDTO):
res = None
async with self.uow:
try:
res = await self.uow.users.patch_one(filter={"id": id}, data=data)
@ -63,14 +61,11 @@ class UserService:
async def delete_user(self, id: UUID) -> None | str:
async with self.uow:
try:
await self.uow.users.delete_one(filter={"id": id})
res = await self.uow.users.delete_one(filter={"id": id})
except IntegrityError as e:
await self.uow.rollback()
res = e._message()
else:
res = None
await self.uow.commit()
finally:
return res
return None