how to query SQLAlchemyRepository methods?

[quick nebula]

Hello, I'm trying to check first if a customer exists in the database using get_one_or_none, but I don't think the current repository has access to the base model.

#controller.py
@post(
        path=urls.REGISTER_CUSTOMER,
        dependencies={"customer_transaction": Provide(provide_customer_service)},
        guards=[requires_account_management_user]
    )
    async def register_customer(self, data: CustomerRegisterSchema, request: Request[EmployeeModel, Token, Any],
                                customer_transaction: CustomerService) -> Response:
        user_data = CustomerModel(
            first_name=data.first_name,
            last_name=data.last_name,
            email=data.email,
            phone=data.phone,
            hashed_password=await hash_password(data.password),
            user_type=UserType[data.user_type],
            company=data.company,
            shortname=data.shortname,
            contact_emails=data.contact_emails,
            address=data.address,
            representative_id=request.user.id
        )

        user = customer_transaction.get_one_or_none(email=data.email) #returns coroutine object
        user = await customer_transaction.get_one_or_none(email=data.email) #returns None

        if user:
            return Response(status_code=400, content="Customer already exists")
        await customer_transaction.create(user_data)

        return Response(status_code=200, content="Customer created successfully")

[stable salmon]

Notes for how to query SQLAlchemyRepository methods?

At your assistance

@quick nebula

No Response?

If no response in a reasonable time, ping @Member.

Closing

To close, type !solve or byte solve.

MCVE

Please include an MCVE so that we can reproduce your issue locally.

[quick nebula]

I think the problem might be the email is not present in the CustomerModel but rather the CustomerModel inherits from the UserModel.

class CustomerModel(UserModel):

    __tablename__ = 'customers_account'

    id: Mapped[UUID] = mapped_column(ForeignKey('users_account.id'), primary_key=True)
    company: Mapped[str] = mapped_column(String, nullable=False)
    shortname: Mapped[str] = mapped_column(String, nullable=True)
    address: Mapped[str] = mapped_column(String, nullable=False)
    contact_emails: Mapped[list[str]] = mapped_column(JSON, nullable=True)
    representative_id: Mapped[UUID] = mapped_column(ForeignKey('employees_account.id'), nullable=False)
    representative: Mapped["EmployeeModel"] = relationship("EmployeeModel", back_populates="customer", foreign_keys=[representative_id])

    __mapper_args__ = {
        'polymorphic_identity': UserType.customer,
    }

class UserModel(UUIDAuditBase):
    __tablename__ = 'users_account'

    first_name: Mapped[str] = mapped_column(nullable=False)
    last_name: Mapped[str] = mapped_column(nullable=False)
    email: Mapped[str] = mapped_column(nullable=False, unique=True)
    phone: Mapped[str] = mapped_column(nullable=False, unique=True)
    hashed_password: Mapped[str] = mapped_column(nullable=False)
    user_type: Mapped[UserType] = mapped_column(nullable=False, default=UserType.guest.value)

    __mapper_args__ = {
        'polymorphic_identity': 'users',
        'polymorphic_on': user_type
    }

This only access the CustomerModel

#database.py
async def provide_customer_repo(db_session: AsyncSession) -> CustomerRepository:
    """This provides the default Customer repository."""
    return CustomerRepository(session=db_session)

I'm not quite sure how to query the email from the UserModel using the customer_transaction which only has access to CustomerModel

[quick nebula]

I think I solved it. I don't know why I added the await here user = await customer_transaction.get_one_or_none(email=data.email), but when I removed it, it shows the object.

I checked the terminal and it also showed a warning:
/app/.venv/lib/python3.12/site-packages/litestar/routes/http.py:202: RuntimeWarning: coroutine 'SQLAlchemyAsyncRepository.get_one_or_none' was never awaited

not sure if this can impact my project in the future, but i really don't want any warnings during runtime

[foggy sand]

its not doing what you think it is

you should await

[quick nebula]

user = customer_transaction.get_one_or_none(email=data.email)

"""
2024-06-04 15:20:22 ******************************
2024-06-04 15:20:22 ******************************
2024-06-04 15:20:22 <coroutine object SQLAlchemyAsyncRepositoryReadService.get_one_or_none at 0x7fc3c115a020>
2024-06-04 15:20:22 ******************************
2024-06-04 15:20:22 ******************************
"""

user = await customer_transaction.get_one_or_none(email=data.email)

"""
2024-06-04 15:15:32 ******************************
2024-06-04 15:15:32 ******************************
2024-06-04 15:15:32 None
2024-06-04 15:15:32 ******************************
2024-06-04 15:15:32 ******************************
"""

Why is the user returning None when I await it?

[quick nebula]

Btw, I modifed this code to this:

async def provide_customer_service(db_session: AsyncSession) -> AsyncGenerator[CustomerService, None]:
    """This provides the default User repository."""
    async with CustomerService.new(session=db_session) as repo:
        yield repo

[foggy sand]

can you show CustomerRepository?

[foggy sand]

this is still wrong

[foggy sand]

with this I can debug

I have an inherited model I use AA on, it works

[quick nebula]

#repository.py
class CustomerRepository(SQLAlchemyAsyncRepository[CustomerModel]):

    model_type = CustomerModel

[foggy sand]

ok, so obvious question, does email exist in the table?

[quick nebula]

yes

when I don't put an await it returns the customer object

[foggy sand]

it does not

can you print data.email?

and also show what is the value in your table

[quick nebula]

POSTMAN
{
  "email": "[email protected],
  ...
}

[foggy sand]

i mean, what is the value for data.email in your code

and what is the value that is in the table

this is not a bug, I have a similar usecase where I filter on the base attribute, so I know this works

[quick nebula]

yeah, i know. I entered it by mistake

value of print(data.email)

2024-06-04 15:36:24 [email protected] <---

[foggy sand]

and this matches the one in the table?

[quick nebula]

yes

[foggy sand]

I think the problem might be the email is not present in the CustomerModel but rather the CustomerModel inherits from the UserModel.

if this were the case it would error out, not return None

[quick nebula]

the logs here is from the non awaited version

let me show you the awaited version

[foggy sand]

not awaiting is not even an option

i understand your point

but what I am suggesting is, you should await

what you think is the object is not the object

if its None, something else is wrong

[quick nebula]

yeah it returns None, but it raise an integrity error

[foggy sand]

ok, is that what you expect?

[quick nebula]

sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) <class 'asyncpg.exceptions.UniqueViolationError'>: duplicate key value violates unique constraint "uq_users_account_email"
DETAIL:  Key (email)=([email protected]) already exists.```

[quick nebula]

Yes, but I expect it to return the response of Customer already exists and not raise an integrity error

[foggy sand]

do you have an MCVE?

[quick nebula]

not yet let me create one for you

[foggy sand]

thanks

[quick nebula]

do you want me to just upload it to github? the whole code?

[foggy sand]

something I can just copy paste

you neednt have to upload the entire thing, something in one file

@quick nebula can you do await customer_transaction.list()

sometime before the get_or_none

[quick nebula]

it's returning empty list

2024-06-04 16:04:34 []

[quick nebula]

Hey @foggy sand, I think I know where the problem is. I am querying the email in CustomerModel, but I'm not sure if that is the proper way of querying the data since the email attribute is at the UserModel, which is inherited by CustomerModel. see this model file

[foggy sand]

.

can you show CustomerService?

[quick nebula]

this is the CustomerService

class CustomerService(SQLAlchemyAsyncRepositoryService[CustomerModel]):
    """Handles database operations for customers."""

    repository_type = CustomerRepository

#model.py
class CustomerModel(UserModel):

    __tablename__ = 'customers_account'

    id: Mapped[UUID] = mapped_column(ForeignKey('users_account.id'), primary_key=True)
    company: Mapped[str] = mapped_column(String, nullable=False)
    shortname: Mapped[str] = mapped_column(String, nullable=True)
    address: Mapped[str] = mapped_column(String, nullable=False)
    contact_emails: Mapped[list[str]] = mapped_column(JSON, nullable=True)
    representative_id: Mapped[UUID] = mapped_column(ForeignKey('employees_account.id'), nullable=False)
    representative: Mapped["EmployeeModel"] = relationship("EmployeeModel", back_populates="customer", foreign_keys=[representative_id])

    __mapper_args__ = {
        'polymorphic_identity': UserType.customer,
    }

#repository.py
class CustomerRepository(SQLAlchemyAsyncRepository[CustomerModel]):

    model_type = CustomerModel

I listed all the relevant codes here so you don't have to scroll back up

[foggy sand]

i am not sure why this is empty

you have shown your table has data, but that is empty

[quick nebula]

I know the problem

[foggy sand]

what is it

[quick nebula]

the email is [email protected] and I am querying under the CustomerModel(UserModel). it shows none because there is no customer1 inside the customermodel. the user type of customer1 is employee

I tried querying it on the EmployeeModel and it returns the model object. verifying it using to_model it shows the correct data

idk why I keep on making these kinds of mistake. 😫

[foggy sand]

help me understand more

why is that list empty

there is no filter there

[foggy sand]

.

it should show the two rows you have right

1 row*

[quick nebula]

yeah it is empty because it is fetching the data from the CustomerModel. when I change it to EmployeeModel it shows the row data

I have an employeemodel i just didnt include it here because I was using customermodel earlier

[foggy sand]

so you have another enum

customer1 is actually going into a third table?

[quick nebula]

no not really. let me show it to you later once I got home

[foggy sand]

thanks

[quick nebula]

earlier, what I am trying to access is the register_customer endpoint. the endpoint should check the data in UserModel using the data.email.

The CustomerModel and EmployeeModel inherit the UserModel.

The error I am receiving earlier is this line of code user = await customer_transaction.get_one_or_none(email=data.email) is returning None because the data.email = [email protected] (which I though was a user_type = customer, but I forgot that I didn't change the user_type and its value is employee. So, basically, I am querying the CustomerModel using the data.email which has a user_type of employee.

I mentioned that it returned None. You said that it does not find it in the table CustomerModel. which is partially correct, but it detected the data, and it raised an IntegrityError:

sqlalchemy.exc.IntegrityError: (sqlalchemy.dialects.postgresql.asyncpg.IntegrityError) <class 'asyncpg.exceptions.UniqueViolationError'>: duplicate key value violates unique constraint "uq_users_account_email"
DETAIL:  Key (email)=([email protected]) already exists.

The correct query should be the EmployeeModel because the [email protected] has a user_type of employee

Test Run #1:

POST /register/employee

Body:
{
    "first_name": "Customer1",
    "last_name": "user",
    "email": "[email protected]",
    "phone": "09000000002",
    "password": "test123!",
    "user_type": "employee",
    "department": "sales"
}

Result: Success 200

Test Run #2:

POST /register/customer

Body:
{
    "first_name": "customer1",
    "last_name": "user",
    "email": "[email protected]",
    "phone": "09000000003",
    "password": "test123!",
    "user_type": "customer",
    "company": "test",
    "contact_emails": ["[email protected]", "[email protected]"],
    "address": ""
}

Result:

• user = await customer_transaction.get_one_or_none(email=data.email) -> `None
• raises advanced_alchemy.exceptions.IntegrityError

hopefully, I explained it properly.

tldr;
I am querying employee data in CustomerModel. The right query should be employee data in EmployeeModel

[foggy sand]

thanks, but in this pic, the email you highlighted has customer as type though

[quick nebula]

I think I made a mistake somewhere. I tried putting it in employee and customer and it worked as expected

[foggy sand]

I mentioned that it returned None. You said that it does not find it in the table CustomerModel. which is partially correct, but it detected the data, and it raised an IntegrityError:

fwiw, it detected it not on customer but on user, cuz thats where email is

you may write the data using the sub class object, but SQLA will do the correct queries

[quick nebula]

yeah, it's my first time using it so it kinda confuses me on how to properly query data

[foggy sand]

thanks for the explanation, appreciate it

mark this as solved then?

[quick nebula]

yep, thanks!

!solve