users.py 31 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899
  1. import json
  2. from datetime import datetime
  3. from typing import Any, Dict, List, Optional
  4. from uuid import UUID
  5. from fastapi import HTTPException
  6. from core.base import CryptoProvider, Handler
  7. from core.base.abstractions import R2RException
  8. from core.utils import generate_user_id
  9. from shared.abstractions import User
  10. from .base import PostgresConnectionManager, QueryBuilder
  11. from .collections import PostgresCollectionsHandler
  12. class PostgresUserHandler(Handler):
  13. TABLE_NAME = "users"
  14. API_KEYS_TABLE_NAME = "users_api_keys"
  15. def __init__(
  16. self,
  17. project_name: str,
  18. connection_manager: PostgresConnectionManager,
  19. crypto_provider: CryptoProvider,
  20. ):
  21. super().__init__(project_name, connection_manager)
  22. self.crypto_provider = crypto_provider
  23. async def create_tables(self):
  24. user_table_query = f"""
  25. CREATE TABLE IF NOT EXISTS {self._get_table_name(PostgresUserHandler.TABLE_NAME)} (
  26. id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
  27. email TEXT UNIQUE NOT NULL,
  28. hashed_password TEXT NOT NULL,
  29. is_superuser BOOLEAN DEFAULT FALSE,
  30. is_active BOOLEAN DEFAULT TRUE,
  31. is_verified BOOLEAN DEFAULT FALSE,
  32. verification_code TEXT,
  33. verification_code_expiry TIMESTAMPTZ,
  34. name TEXT,
  35. bio TEXT,
  36. profile_picture TEXT,
  37. reset_token TEXT,
  38. reset_token_expiry TIMESTAMPTZ,
  39. collection_ids UUID[] NULL,
  40. limits_overrides JSONB,
  41. created_at TIMESTAMPTZ DEFAULT NOW(),
  42. updated_at TIMESTAMPTZ DEFAULT NOW()
  43. );
  44. """
  45. # API keys table with updated_at instead of last_used_at
  46. api_keys_table_query = f"""
  47. CREATE TABLE IF NOT EXISTS {self._get_table_name(PostgresUserHandler.API_KEYS_TABLE_NAME)} (
  48. id UUID PRIMARY KEY DEFAULT uuid_generate_v4(),
  49. user_id UUID NOT NULL REFERENCES {self._get_table_name(PostgresUserHandler.TABLE_NAME)}(id) ON DELETE CASCADE,
  50. public_key TEXT UNIQUE NOT NULL,
  51. hashed_key TEXT NOT NULL,
  52. name TEXT,
  53. created_at TIMESTAMPTZ DEFAULT NOW(),
  54. updated_at TIMESTAMPTZ DEFAULT NOW()
  55. );
  56. CREATE INDEX IF NOT EXISTS idx_api_keys_user_id
  57. ON {self._get_table_name(PostgresUserHandler.API_KEYS_TABLE_NAME)}(user_id);
  58. CREATE INDEX IF NOT EXISTS idx_api_keys_public_key
  59. ON {self._get_table_name(PostgresUserHandler.API_KEYS_TABLE_NAME)}(public_key);
  60. """
  61. await self.connection_manager.execute_query(user_table_query)
  62. await self.connection_manager.execute_query(api_keys_table_query)
  63. async def get_user_by_id(self, id: UUID) -> User:
  64. query, _ = (
  65. QueryBuilder(self._get_table_name("users"))
  66. .select(
  67. [
  68. "id",
  69. "email",
  70. "hashed_password",
  71. "is_superuser",
  72. "is_active",
  73. "is_verified",
  74. "created_at",
  75. "updated_at",
  76. "name",
  77. "profile_picture",
  78. "bio",
  79. "collection_ids",
  80. "limits_overrides", # Fetch JSONB column
  81. ]
  82. )
  83. .where("id = $1")
  84. .build()
  85. )
  86. result = await self.connection_manager.fetchrow_query(query, [id])
  87. if not result:
  88. raise R2RException(status_code=404, message="User not found")
  89. return User(
  90. id=result["id"],
  91. email=result["email"],
  92. hashed_password=result["hashed_password"],
  93. is_superuser=result["is_superuser"],
  94. is_active=result["is_active"],
  95. is_verified=result["is_verified"],
  96. created_at=result["created_at"],
  97. updated_at=result["updated_at"],
  98. name=result["name"],
  99. profile_picture=result["profile_picture"],
  100. bio=result["bio"],
  101. collection_ids=result["collection_ids"],
  102. # Add the new field
  103. limits_overrides=json.loads(result["limits_overrides"] or "{}"),
  104. )
  105. async def get_user_by_email(self, email: str) -> User:
  106. query, params = (
  107. QueryBuilder(self._get_table_name("users"))
  108. .select(
  109. [
  110. "id",
  111. "email",
  112. "hashed_password",
  113. "is_superuser",
  114. "is_active",
  115. "is_verified",
  116. "created_at",
  117. "updated_at",
  118. "name",
  119. "profile_picture",
  120. "bio",
  121. "collection_ids",
  122. "limits_overrides",
  123. ]
  124. )
  125. .where("email = $1")
  126. .build()
  127. )
  128. result = await self.connection_manager.fetchrow_query(query, [email])
  129. if not result:
  130. raise R2RException(status_code=404, message="User not found")
  131. return User(
  132. id=result["id"],
  133. email=result["email"],
  134. hashed_password=result["hashed_password"],
  135. is_superuser=result["is_superuser"],
  136. is_active=result["is_active"],
  137. is_verified=result["is_verified"],
  138. created_at=result["created_at"],
  139. updated_at=result["updated_at"],
  140. name=result["name"],
  141. profile_picture=result["profile_picture"],
  142. bio=result["bio"],
  143. collection_ids=result["collection_ids"],
  144. limits_overrides=json.loads(result["limits_overrides"] or "{}"),
  145. )
  146. async def create_user(
  147. self, email: str, password: str, is_superuser: bool = False
  148. ) -> User:
  149. """Create a new user."""
  150. try:
  151. existing = await self.get_user_by_email(email)
  152. if existing:
  153. raise R2RException(
  154. status_code=400,
  155. message="User with this email already exists",
  156. )
  157. except R2RException as e:
  158. if e.status_code != 404:
  159. raise e
  160. hashed_password = self.crypto_provider.get_password_hash(password) # type: ignore
  161. query, params = (
  162. QueryBuilder(self._get_table_name(self.TABLE_NAME))
  163. .insert(
  164. {
  165. "email": email,
  166. "id": generate_user_id(email),
  167. "is_superuser": is_superuser,
  168. "hashed_password": hashed_password,
  169. "collection_ids": [],
  170. "limits_overrides": None,
  171. }
  172. )
  173. .returning(
  174. [
  175. "id",
  176. "email",
  177. "is_superuser",
  178. "is_active",
  179. "is_verified",
  180. "created_at",
  181. "updated_at",
  182. "collection_ids",
  183. "limits_overrides",
  184. ]
  185. )
  186. .build()
  187. )
  188. result = await self.connection_manager.fetchrow_query(query, params)
  189. if not result:
  190. raise R2RException(
  191. status_code=500,
  192. message="Failed to create user",
  193. )
  194. return User(
  195. id=result["id"],
  196. email=result["email"],
  197. is_superuser=result["is_superuser"],
  198. is_active=result["is_active"],
  199. is_verified=result["is_verified"],
  200. created_at=result["created_at"],
  201. updated_at=result["updated_at"],
  202. collection_ids=result["collection_ids"] or [],
  203. hashed_password=hashed_password,
  204. limits_overrides=json.loads(result["limits_overrides"] or "{}"),
  205. name=None,
  206. bio=None,
  207. profile_picture=None,
  208. )
  209. async def update_user(
  210. self, user: User, merge_limits: bool = False
  211. ) -> User:
  212. """
  213. Update user information including limits_overrides.
  214. Args:
  215. user: User object containing updated information
  216. merge_limits: If True, will merge existing limits_overrides with new ones.
  217. If False, will overwrite existing limits_overrides.
  218. Returns:
  219. Updated User object
  220. """
  221. # Get current user if we need to merge limits or get hashed password
  222. current_user = None
  223. try:
  224. current_user = await self.get_user_by_id(user.id)
  225. except R2RException:
  226. raise R2RException(status_code=404, message="User not found")
  227. # Merge or replace limits_overrides
  228. final_limits = user.limits_overrides
  229. if (
  230. merge_limits
  231. and current_user.limits_overrides
  232. and user.limits_overrides
  233. ):
  234. final_limits = {
  235. **current_user.limits_overrides,
  236. **user.limits_overrides,
  237. }
  238. query = f"""
  239. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  240. SET email = $1,
  241. is_superuser = $2,
  242. is_active = $3,
  243. is_verified = $4,
  244. updated_at = NOW(),
  245. name = $5,
  246. profile_picture = $6,
  247. bio = $7,
  248. collection_ids = $8,
  249. limits_overrides = $9::jsonb
  250. WHERE id = $10
  251. RETURNING id, email, is_superuser, is_active, is_verified,
  252. created_at, updated_at, name, profile_picture, bio,
  253. collection_ids, limits_overrides, hashed_password
  254. """
  255. result = await self.connection_manager.fetchrow_query(
  256. query,
  257. [
  258. user.email,
  259. user.is_superuser,
  260. user.is_active,
  261. user.is_verified,
  262. user.name,
  263. user.profile_picture,
  264. user.bio,
  265. user.collection_ids or [], # Ensure null becomes empty array
  266. json.dumps(final_limits), # Already handled null case
  267. user.id,
  268. ],
  269. )
  270. if not result:
  271. raise HTTPException(
  272. status_code=500,
  273. detail="Failed to update user",
  274. )
  275. return User(
  276. id=result["id"],
  277. email=result["email"],
  278. hashed_password=result[
  279. "hashed_password"
  280. ], # Include hashed_password
  281. is_superuser=result["is_superuser"],
  282. is_active=result["is_active"],
  283. is_verified=result["is_verified"],
  284. created_at=result["created_at"],
  285. updated_at=result["updated_at"],
  286. name=result["name"],
  287. profile_picture=result["profile_picture"],
  288. bio=result["bio"],
  289. collection_ids=result["collection_ids"]
  290. or [], # Ensure null becomes empty array
  291. limits_overrides=json.loads(
  292. result["limits_overrides"] or "{}"
  293. ), # Can be null
  294. )
  295. async def delete_user_relational(self, id: UUID) -> None:
  296. """Delete a user and update related records."""
  297. # Get the collections the user belongs to
  298. collection_query, params = (
  299. QueryBuilder(self._get_table_name(self.TABLE_NAME))
  300. .select(["collection_ids"])
  301. .where("id = $1")
  302. .build()
  303. )
  304. collection_result = await self.connection_manager.fetchrow_query(
  305. collection_query, [id]
  306. )
  307. if not collection_result:
  308. raise R2RException(status_code=404, message="User not found")
  309. # Update documents query
  310. doc_update_query, doc_params = (
  311. QueryBuilder(self._get_table_name("documents"))
  312. .update({"id": None})
  313. .where("id = $1")
  314. .build()
  315. )
  316. await self.connection_manager.execute_query(doc_update_query, [id])
  317. # Delete user query
  318. delete_query, del_params = (
  319. QueryBuilder(self._get_table_name(self.TABLE_NAME))
  320. .delete()
  321. .where("id = $1")
  322. .returning(["id"])
  323. .build()
  324. )
  325. result = await self.connection_manager.fetchrow_query(
  326. delete_query, [id]
  327. )
  328. if not result:
  329. raise R2RException(status_code=404, message="User not found")
  330. async def update_user_password(self, id: UUID, new_hashed_password: str):
  331. query = f"""
  332. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  333. SET hashed_password = $1, updated_at = NOW()
  334. WHERE id = $2
  335. """
  336. await self.connection_manager.execute_query(
  337. query, [new_hashed_password, id]
  338. )
  339. async def get_all_users(self) -> List[User]:
  340. """Get all users with minimal information."""
  341. query, params = (
  342. QueryBuilder(self._get_table_name(self.TABLE_NAME))
  343. .select(
  344. [
  345. "id",
  346. "email",
  347. "is_superuser",
  348. "is_active",
  349. "is_verified",
  350. "created_at",
  351. "updated_at",
  352. "collection_ids",
  353. "hashed_password",
  354. "limits_overrides",
  355. "name",
  356. "bio",
  357. "profile_picture",
  358. ]
  359. )
  360. .build()
  361. )
  362. results = await self.connection_manager.fetch_query(query, params)
  363. return [
  364. User(
  365. id=result["id"],
  366. email=result["email"],
  367. hashed_password=result["hashed_password"],
  368. is_superuser=result["is_superuser"],
  369. is_active=result["is_active"],
  370. is_verified=result["is_verified"],
  371. created_at=result["created_at"],
  372. updated_at=result["updated_at"],
  373. collection_ids=result["collection_ids"] or [],
  374. limits_overrides=json.loads(
  375. result["limits_overrides"] or "{}"
  376. ),
  377. name=result["name"],
  378. bio=result["bio"],
  379. profile_picture=result["profile_picture"],
  380. )
  381. for result in results
  382. ]
  383. async def store_verification_code(
  384. self, id: UUID, verification_code: str, expiry: datetime
  385. ):
  386. query = f"""
  387. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  388. SET verification_code = $1, verification_code_expiry = $2
  389. WHERE id = $3
  390. """
  391. await self.connection_manager.execute_query(
  392. query, [verification_code, expiry, id]
  393. )
  394. async def verify_user(self, verification_code: str) -> None:
  395. query = f"""
  396. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  397. SET is_verified = TRUE, verification_code = NULL, verification_code_expiry = NULL
  398. WHERE verification_code = $1 AND verification_code_expiry > NOW()
  399. RETURNING id
  400. """
  401. result = await self.connection_manager.fetchrow_query(
  402. query, [verification_code]
  403. )
  404. if not result:
  405. raise R2RException(
  406. status_code=400, message="Invalid or expired verification code"
  407. )
  408. async def remove_verification_code(self, verification_code: str):
  409. query = f"""
  410. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  411. SET verification_code = NULL, verification_code_expiry = NULL
  412. WHERE verification_code = $1
  413. """
  414. await self.connection_manager.execute_query(query, [verification_code])
  415. async def expire_verification_code(self, id: UUID):
  416. query = f"""
  417. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  418. SET verification_code_expiry = NOW() - INTERVAL '1 day'
  419. WHERE id = $1
  420. """
  421. await self.connection_manager.execute_query(query, [id])
  422. async def store_reset_token(
  423. self, id: UUID, reset_token: str, expiry: datetime
  424. ):
  425. query = f"""
  426. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  427. SET reset_token = $1, reset_token_expiry = $2
  428. WHERE id = $3
  429. """
  430. await self.connection_manager.execute_query(
  431. query, [reset_token, expiry, id]
  432. )
  433. async def get_user_id_by_reset_token(
  434. self, reset_token: str
  435. ) -> Optional[UUID]:
  436. query = f"""
  437. SELECT id FROM {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  438. WHERE reset_token = $1 AND reset_token_expiry > NOW()
  439. """
  440. result = await self.connection_manager.fetchrow_query(
  441. query, [reset_token]
  442. )
  443. return result["id"] if result else None
  444. async def remove_reset_token(self, id: UUID):
  445. query = f"""
  446. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  447. SET reset_token = NULL, reset_token_expiry = NULL
  448. WHERE id = $1
  449. """
  450. await self.connection_manager.execute_query(query, [id])
  451. async def remove_user_from_all_collections(self, id: UUID):
  452. query = f"""
  453. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  454. SET collection_ids = ARRAY[]::UUID[]
  455. WHERE id = $1
  456. """
  457. await self.connection_manager.execute_query(query, [id])
  458. async def add_user_to_collection(
  459. self, id: UUID, collection_id: UUID
  460. ) -> bool:
  461. # Check if the user exists
  462. if not await self.get_user_by_id(id):
  463. raise R2RException(status_code=404, message="User not found")
  464. # Check if the collection exists
  465. if not await self._collection_exists(collection_id):
  466. raise R2RException(status_code=404, message="Collection not found")
  467. query = f"""
  468. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  469. SET collection_ids = array_append(collection_ids, $1)
  470. WHERE id = $2 AND NOT ($1 = ANY(collection_ids))
  471. RETURNING id
  472. """
  473. result = await self.connection_manager.fetchrow_query(
  474. query, [collection_id, id]
  475. )
  476. if not result:
  477. raise R2RException(
  478. status_code=400, message="User already in collection"
  479. )
  480. update_collection_query = f"""
  481. UPDATE {self._get_table_name('collections')}
  482. SET user_count = user_count + 1
  483. WHERE id = $1
  484. """
  485. await self.connection_manager.execute_query(
  486. query=update_collection_query,
  487. params=[collection_id],
  488. )
  489. return True
  490. async def remove_user_from_collection(
  491. self, id: UUID, collection_id: UUID
  492. ) -> bool:
  493. if not await self.get_user_by_id(id):
  494. raise R2RException(status_code=404, message="User not found")
  495. query = f"""
  496. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  497. SET collection_ids = array_remove(collection_ids, $1)
  498. WHERE id = $2 AND $1 = ANY(collection_ids)
  499. RETURNING id
  500. """
  501. result = await self.connection_manager.fetchrow_query(
  502. query, [collection_id, id]
  503. )
  504. if not result:
  505. raise R2RException(
  506. status_code=400,
  507. message="User is not a member of the specified collection",
  508. )
  509. return True
  510. async def get_users_in_collection(
  511. self, collection_id: UUID, offset: int, limit: int
  512. ) -> dict[str, list[User] | int]:
  513. """Get all users in a specific collection with pagination."""
  514. if not await self._collection_exists(collection_id):
  515. raise R2RException(status_code=404, message="Collection not found")
  516. query, params = (
  517. QueryBuilder(self._get_table_name(self.TABLE_NAME))
  518. .select(
  519. [
  520. "id",
  521. "email",
  522. "is_active",
  523. "is_superuser",
  524. "created_at",
  525. "updated_at",
  526. "is_verified",
  527. "collection_ids",
  528. "name",
  529. "bio",
  530. "profile_picture",
  531. "hashed_password",
  532. "limits_overrides",
  533. "COUNT(*) OVER() AS total_entries",
  534. ]
  535. )
  536. .where("$1 = ANY(collection_ids)")
  537. .order_by("name")
  538. .offset("$2")
  539. .limit("$3" if limit != -1 else None)
  540. .build()
  541. )
  542. conditions = [collection_id, offset]
  543. if limit != -1:
  544. conditions.append(limit)
  545. results = await self.connection_manager.fetch_query(query, conditions)
  546. users_list = [
  547. User(
  548. id=row["id"],
  549. email=row["email"],
  550. is_active=row["is_active"],
  551. is_superuser=row["is_superuser"],
  552. created_at=row["created_at"],
  553. updated_at=row["updated_at"],
  554. is_verified=row["is_verified"],
  555. collection_ids=row["collection_ids"] or [],
  556. name=row["name"],
  557. bio=row["bio"],
  558. profile_picture=row["profile_picture"],
  559. hashed_password=row["hashed_password"],
  560. limits_overrides=json.loads(row["limits_overrides"] or "{}"),
  561. )
  562. for row in results
  563. ]
  564. total_entries = results[0]["total_entries"] if results else 0
  565. return {"results": users_list, "total_entries": total_entries}
  566. async def mark_user_as_superuser(self, id: UUID):
  567. query = f"""
  568. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  569. SET is_superuser = TRUE, is_verified = TRUE,
  570. verification_code = NULL, verification_code_expiry = NULL
  571. WHERE id = $1
  572. """
  573. await self.connection_manager.execute_query(query, [id])
  574. async def get_user_id_by_verification_code(
  575. self, verification_code: str
  576. ) -> UUID:
  577. query = f"""
  578. SELECT id FROM {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  579. WHERE verification_code = $1 AND verification_code_expiry > NOW()
  580. """
  581. result = await self.connection_manager.fetchrow_query(
  582. query, [verification_code]
  583. )
  584. if not result:
  585. raise R2RException(
  586. status_code=400, message="Invalid or expired verification code"
  587. )
  588. return result["id"]
  589. async def mark_user_as_verified(self, id: UUID):
  590. query = f"""
  591. UPDATE {self._get_table_name(PostgresUserHandler.TABLE_NAME)}
  592. SET is_verified = TRUE,
  593. verification_code = NULL,
  594. verification_code_expiry = NULL
  595. WHERE id = $1
  596. """
  597. await self.connection_manager.execute_query(query, [id])
  598. async def get_users_overview(
  599. self,
  600. offset: int,
  601. limit: int,
  602. user_ids: Optional[list[UUID]] = None,
  603. ) -> dict[str, list[User] | int]:
  604. """
  605. Return users with document usage and total entries.
  606. """
  607. query = f"""
  608. WITH user_document_ids AS (
  609. SELECT
  610. u.id as user_id,
  611. ARRAY_AGG(d.id) FILTER (WHERE d.id IS NOT NULL) AS doc_ids
  612. FROM {self._get_table_name(PostgresUserHandler.TABLE_NAME)} u
  613. LEFT JOIN {self._get_table_name('documents')} d ON u.id = d.owner_id
  614. GROUP BY u.id
  615. ),
  616. user_docs AS (
  617. SELECT
  618. u.id,
  619. u.email,
  620. u.is_superuser,
  621. u.is_active,
  622. u.is_verified,
  623. u.name,
  624. u.bio,
  625. u.profile_picture,
  626. u.collection_ids,
  627. u.created_at,
  628. u.updated_at,
  629. COUNT(d.id) AS num_files,
  630. COALESCE(SUM(d.size_in_bytes), 0) AS total_size_in_bytes,
  631. ud.doc_ids as document_ids
  632. FROM {self._get_table_name(PostgresUserHandler.TABLE_NAME)} u
  633. LEFT JOIN {self._get_table_name('documents')} d ON u.id = d.owner_id
  634. LEFT JOIN user_document_ids ud ON u.id = ud.user_id
  635. {' WHERE u.id = ANY($3::uuid[])' if user_ids else ''}
  636. GROUP BY u.id, u.email, u.is_superuser, u.is_active, u.is_verified,
  637. u.created_at, u.updated_at, u.collection_ids, ud.doc_ids
  638. )
  639. SELECT
  640. user_docs.*,
  641. COUNT(*) OVER() AS total_entries
  642. FROM user_docs
  643. ORDER BY email
  644. OFFSET $1
  645. """
  646. params: list = [offset]
  647. if limit != -1:
  648. query += " LIMIT $2"
  649. params.append(limit)
  650. if user_ids:
  651. params.append(user_ids)
  652. results = await self.connection_manager.fetch_query(query, params)
  653. if not results:
  654. raise R2RException(status_code=404, message="No users found")
  655. users_list = []
  656. for row in results:
  657. users_list.append(
  658. User(
  659. id=row["id"],
  660. email=row["email"],
  661. is_superuser=row["is_superuser"],
  662. is_active=row["is_active"],
  663. is_verified=row["is_verified"],
  664. name=row["name"],
  665. bio=row["bio"],
  666. created_at=row["created_at"],
  667. updated_at=row["updated_at"],
  668. profile_picture=row["profile_picture"],
  669. collection_ids=row["collection_ids"] or [],
  670. num_files=row["num_files"],
  671. total_size_in_bytes=row["total_size_in_bytes"],
  672. document_ids=(
  673. list(row["document_ids"])
  674. if row["document_ids"]
  675. else []
  676. ),
  677. )
  678. )
  679. total_entries = results[0]["total_entries"]
  680. return {"results": users_list, "total_entries": total_entries}
  681. async def _collection_exists(self, collection_id: UUID) -> bool:
  682. """Check if a collection exists."""
  683. query = f"""
  684. SELECT 1 FROM {self._get_table_name(PostgresCollectionsHandler.TABLE_NAME)}
  685. WHERE id = $1
  686. """
  687. result = await self.connection_manager.fetchrow_query(
  688. query, [collection_id]
  689. )
  690. return result is not None
  691. async def get_user_validation_data(
  692. self,
  693. user_id: UUID,
  694. ) -> dict:
  695. """
  696. Get verification data for a specific user.
  697. This method should be called after superuser authorization has been verified.
  698. """
  699. query = f"""
  700. SELECT
  701. verification_code,
  702. verification_code_expiry,
  703. reset_token,
  704. reset_token_expiry
  705. FROM {self._get_table_name("users")}
  706. WHERE id = $1
  707. """
  708. result = await self.connection_manager.fetchrow_query(query, [user_id])
  709. if not result:
  710. raise R2RException(status_code=404, message="User not found")
  711. return {
  712. "verification_data": {
  713. "verification_code": result["verification_code"],
  714. "verification_code_expiry": (
  715. result["verification_code_expiry"].isoformat()
  716. if result["verification_code_expiry"]
  717. else None
  718. ),
  719. "reset_token": result["reset_token"],
  720. "reset_token_expiry": (
  721. result["reset_token_expiry"].isoformat()
  722. if result["reset_token_expiry"]
  723. else None
  724. ),
  725. }
  726. }
  727. # API Key methods
  728. async def store_user_api_key(
  729. self,
  730. user_id: UUID,
  731. key_id: str,
  732. hashed_key: str,
  733. name: Optional[str] = None,
  734. ) -> UUID:
  735. """Store a new API key for a user."""
  736. query = f"""
  737. INSERT INTO {self._get_table_name(PostgresUserHandler.API_KEYS_TABLE_NAME)}
  738. (user_id, public_key, hashed_key, name)
  739. VALUES ($1, $2, $3, $4)
  740. RETURNING id
  741. """
  742. result = await self.connection_manager.fetchrow_query(
  743. query, [user_id, key_id, hashed_key, name]
  744. )
  745. if not result:
  746. raise R2RException(
  747. status_code=500, message="Failed to store API key"
  748. )
  749. return result["id"]
  750. async def get_api_key_record(self, key_id: str) -> Optional[dict]:
  751. """
  752. Get API key record by 'public_key' and update 'updated_at' to now.
  753. Returns { "user_id", "hashed_key" } or None if not found.
  754. """
  755. query = f"""
  756. UPDATE {self._get_table_name(PostgresUserHandler.API_KEYS_TABLE_NAME)}
  757. SET updated_at = NOW()
  758. WHERE public_key = $1
  759. RETURNING user_id, hashed_key
  760. """
  761. result = await self.connection_manager.fetchrow_query(query, [key_id])
  762. if not result:
  763. return None
  764. return {
  765. "user_id": result["user_id"],
  766. "hashed_key": result["hashed_key"],
  767. }
  768. async def get_user_api_keys(self, user_id: UUID) -> list[dict]:
  769. """Get all API keys for a user."""
  770. query = f"""
  771. SELECT id, public_key, name, created_at, updated_at
  772. FROM {self._get_table_name(PostgresUserHandler.API_KEYS_TABLE_NAME)}
  773. WHERE user_id = $1
  774. ORDER BY created_at DESC
  775. """
  776. results = await self.connection_manager.fetch_query(query, [user_id])
  777. return [
  778. {
  779. "key_id": str(row["id"]),
  780. "public_key": row["public_key"],
  781. "name": row["name"] or "",
  782. "updated_at": row["updated_at"],
  783. }
  784. for row in results
  785. ]
  786. async def delete_api_key(self, user_id: UUID, key_id: UUID) -> dict:
  787. """Delete a specific API key."""
  788. query = f"""
  789. DELETE FROM {self._get_table_name(PostgresUserHandler.API_KEYS_TABLE_NAME)}
  790. WHERE id = $1 AND user_id = $2
  791. RETURNING id, public_key, name
  792. """
  793. result = await self.connection_manager.fetchrow_query(
  794. query, [key_id, user_id]
  795. )
  796. if result is None:
  797. raise R2RException(status_code=404, message="API key not found")
  798. return {
  799. "key_id": str(result["id"]),
  800. "public_key": str(result["public_key"]),
  801. "name": result["name"] or "",
  802. }
  803. async def update_api_key_name(
  804. self, user_id: UUID, key_id: UUID, name: str
  805. ) -> bool:
  806. """Update the name of an existing API key."""
  807. query = f"""
  808. UPDATE {self._get_table_name(PostgresUserHandler.API_KEYS_TABLE_NAME)}
  809. SET name = $1, updated_at = NOW()
  810. WHERE id = $2 AND user_id = $3
  811. RETURNING id
  812. """
  813. result = await self.connection_manager.fetchrow_query(
  814. query, [name, key_id, user_id]
  815. )
  816. if result is None:
  817. raise R2RException(status_code=404, message="API key not found")
  818. return True