Coverage for polar/metrics/metrics.py: 98%

1from collections import deque 

2from collections.abc import Iterable 

3from datetime import datetime 

4from enum import StrEnum 

5from typing import TYPE_CHECKING, ClassVar, Protocol, cast 

6 

7if TYPE_CHECKING: 7 ↛ 8line 7 didn't jump to line 8 because the condition on line 7 was never true

8 from .schemas import MetricsPeriod 

9 

10from sqlalchemy import ( 

11 ColumnElement, 

12 Float, 

13 Integer, 

14 SQLColumnExpression, 

15 case, 

16 func, 

17 or_, 

18 type_coerce, 

19) 

20 

21from polar.enums import SubscriptionRecurringInterval 

22from polar.kit.time_queries import TimeInterval 

23from polar.models import Checkout, Order, Subscription 

24from polar.models.checkout import CheckoutStatus 

25from polar.models.event import Event 

26from polar.models.subscription import CustomerCancellationReason 

27 

28from .queries import MetricQuery 

29 

30 

31class MetricType(StrEnum): 

32 scalar = "scalar" 

33 currency = "currency" 

34 currency_sub_cent = "currency_sub_cent" 

35 percentage = "percentage" 

36 

37 

38def cumulative_sum(periods: Iterable["MetricsPeriod"], slug: str) -> int | float: 

39 return sum(getattr(p, slug) for p in periods) 

40 

41 

42def cumulative_last(periods: Iterable["MetricsPeriod"], slug: str) -> int | float: 

43 dd = deque((getattr(p, slug) for p in periods), maxlen=1) 

44 return dd.pop() 

45 

46 

47class Metric(Protocol): 

48 slug: ClassVar[str] 

49 display_name: ClassVar[str] 

50 type: ClassVar[MetricType] 

51 

52 @classmethod 

53 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: ... 53 ↛ anywhereline 53 didn't jump anywhere: it always raised an exception.

54 

55 

56class SQLMetric(Metric, Protocol): 

57 query: ClassVar[MetricQuery] 

58 

59 @classmethod 

60 def get_sql_expression( 60 ↛ anywhereline 60 didn't jump anywhere: it always raised an exception.

61 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

62 ) -> ColumnElement[int] | ColumnElement[float]: ... 

63 

64 

65class MetaMetric(Metric, Protocol): 

66 @classmethod 

67 def compute_from_period(cls, period: "MetricsPeriod") -> int | float: ... 67 ↛ exitline 67 didn't return from function 'compute_from_period' because

68 

69 

70class OrdersMetric(SQLMetric): 

71 slug = "orders" 

72 display_name = "Orders" 

73 type = MetricType.scalar 

74 query = MetricQuery.orders 

75 

76 @classmethod 

77 def get_sql_expression( 

78 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

79 ) -> ColumnElement[int]: 

80 return func.count(Order.id) 

81 

82 @classmethod 

83 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

84 return cumulative_sum(periods, cls.slug) 

85 

86 

87class RevenueMetric(SQLMetric): 

88 slug = "revenue" 

89 display_name = "Revenue" 

90 type = MetricType.currency 

91 query = MetricQuery.orders 

92 

93 @classmethod 

94 def get_sql_expression( 

95 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

96 ) -> ColumnElement[int]: 

97 return func.sum(Order.net_amount) 

98 

99 @classmethod 

100 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

101 return cumulative_sum(periods, cls.slug) 

102 

103 

104class NetRevenueMetric(SQLMetric): 

105 slug = "net_revenue" 

106 display_name = "Net Revenue" 

107 type = MetricType.currency 

108 query = MetricQuery.orders 

109 

110 @classmethod 

111 def get_sql_expression( 

112 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

113 ) -> ColumnElement[int]: 

114 return func.sum(Order.payout_amount) 

115 

116 @classmethod 

117 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

118 return cumulative_sum(periods, cls.slug) 

119 

120 

121class CumulativeRevenueMetric(SQLMetric): 

122 slug = "cumulative_revenue" 

123 display_name = "Cumulative Revenue" 

124 type = MetricType.currency 

125 query = MetricQuery.orders 

126 

127 @classmethod 

128 def get_sql_expression( 

129 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

130 ) -> ColumnElement[int]: 

131 return func.sum(Order.net_amount) 

132 

133 @classmethod 

134 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

135 return cumulative_last(periods, cls.slug) 

136 

137 

138class NetCumulativeRevenueMetric(SQLMetric): 

139 slug = "net_cumulative_revenue" 

140 display_name = "Net Cumulative Revenue" 

141 type = MetricType.currency 

142 query = MetricQuery.orders 

143 

144 @classmethod 

145 def get_sql_expression( 

146 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

147 ) -> ColumnElement[int]: 

148 return func.sum(Order.payout_amount) 

149 

150 @classmethod 

151 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

152 return cumulative_last(periods, cls.slug) 

153 

154 

155class AverageOrderValueMetric(SQLMetric): 

156 slug = "average_order_value" 

157 display_name = "Average Order Value" 

158 type = MetricType.currency 

159 query = MetricQuery.orders 

160 

161 @classmethod 

162 def get_sql_expression( 

163 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

164 ) -> ColumnElement[int]: 

165 return func.cast(func.ceil(func.avg(Order.net_amount)), Integer) 

166 

167 @classmethod 

168 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 

169 total_orders = sum(getattr(p, "orders") for p in periods) 

170 revenue = sum(getattr(p, "revenue") for p in periods) 

171 return revenue / total_orders if total_orders > 0 else 0.0 

172 

173 

174class NetAverageOrderValueMetric(SQLMetric): 

175 slug = "net_average_order_value" 

176 display_name = "Net Average Order Value" 

177 type = MetricType.currency 

178 query = MetricQuery.orders 

179 

180 @classmethod 

181 def get_sql_expression( 

182 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

183 ) -> ColumnElement[int]: 

184 return func.cast(func.ceil(func.avg(Order.payout_amount)), Integer) 

185 

186 @classmethod 

187 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 

188 total_orders = sum(getattr(p, "orders") for p in periods) 

189 revenue = sum(getattr(p, "net_revenue") for p in periods) 

190 return revenue / total_orders if total_orders > 0 else 0.0 

191 

192 

193class OneTimeProductsMetric(SQLMetric): 

194 slug = "one_time_products" 

195 display_name = "One-Time Products" 

196 type = MetricType.scalar 

197 query = MetricQuery.orders 

198 

199 @classmethod 

200 def get_sql_expression( 

201 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

202 ) -> ColumnElement[int]: 

203 return func.count(Order.id).filter(Order.subscription_id.is_(None)) 

204 

205 @classmethod 

206 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

207 return cumulative_sum(periods, cls.slug) 

208 

209 

210class OneTimeProductsRevenueMetric(SQLMetric): 

211 slug = "one_time_products_revenue" 

212 display_name = "One-Time Products Revenue" 

213 type = MetricType.currency 

214 query = MetricQuery.orders 

215 

216 @classmethod 

217 def get_sql_expression( 

218 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

219 ) -> ColumnElement[int]: 

220 return func.sum(Order.net_amount).filter(Order.subscription_id.is_(None)) 

221 

222 @classmethod 

223 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

224 return cumulative_sum(periods, cls.slug) 

225 

226 

227class OneTimeProductsNetRevenueMetric(SQLMetric): 

228 slug = "one_time_products_net_revenue" 

229 display_name = "One-Time Products Net Revenue" 

230 type = MetricType.currency 

231 query = MetricQuery.orders 

232 

233 @classmethod 

234 def get_sql_expression( 

235 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

236 ) -> ColumnElement[int]: 

237 return func.sum(Order.payout_amount).filter(Order.subscription_id.is_(None)) 

238 

239 @classmethod 

240 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

241 return cumulative_sum(periods, cls.slug) 

242 

243 

244class NewSubscriptionsMetric(SQLMetric): 

245 slug = "new_subscriptions" 

246 display_name = "New Subscriptions" 

247 type = MetricType.scalar 

248 query = MetricQuery.active_subscriptions 

249 

250 @classmethod 

251 def get_sql_expression( 

252 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

253 ) -> ColumnElement[int]: 

254 return func.count(Subscription.id).filter( 

255 i.sql_date_trunc( 

256 cast(SQLColumnExpression[datetime], Subscription.started_at) 

257 ) 

258 == i.sql_date_trunc(t) 

259 ) 

260 

261 @classmethod 

262 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

263 return cumulative_sum(periods, cls.slug) 

264 

265 

266class NewSubscriptionsRevenueMetric(SQLMetric): 

267 slug = "new_subscriptions_revenue" 

268 display_name = "New Subscriptions Revenue" 

269 type = MetricType.currency 

270 query = MetricQuery.orders 

271 

272 @classmethod 

273 def get_sql_expression( 

274 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

275 ) -> ColumnElement[int]: 

276 return func.sum(Order.net_amount).filter( 

277 i.sql_date_trunc( 

278 cast(SQLColumnExpression[datetime], Subscription.started_at) 

279 ) 

280 == i.sql_date_trunc(t) 

281 ) 

282 

283 @classmethod 

284 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

285 return cumulative_sum(periods, cls.slug) 

286 

287 

288class NewSubscriptionsNetRevenueMetric(SQLMetric): 

289 slug = "new_subscriptions_net_revenue" 

290 display_name = "New Subscriptions Net Revenue" 

291 type = MetricType.currency 

292 query = MetricQuery.orders 

293 

294 @classmethod 

295 def get_sql_expression( 

296 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

297 ) -> ColumnElement[int]: 

298 return func.sum(Order.payout_amount).filter( 

299 i.sql_date_trunc( 

300 cast(SQLColumnExpression[datetime], Subscription.started_at) 

301 ) 

302 == i.sql_date_trunc(t) 

303 ) 

304 

305 @classmethod 

306 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

307 return cumulative_sum(periods, cls.slug) 

308 

309 

310class RenewedSubscriptionsMetric(SQLMetric): 

311 slug = "renewed_subscriptions" 

312 display_name = "Renewed Subscriptions" 

313 type = MetricType.scalar 

314 query = MetricQuery.orders 

315 

316 @classmethod 

317 def get_sql_expression( 

318 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

319 ) -> ColumnElement[int]: 

320 return func.count(Subscription.id.distinct()).filter( 

321 i.sql_date_trunc( 

322 cast(SQLColumnExpression[datetime], Subscription.started_at) 

323 ) 

324 != i.sql_date_trunc(t) 

325 ) 

326 

327 @classmethod 

328 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

329 return cumulative_sum(periods, cls.slug) 

330 

331 

332class RenewedSubscriptionsRevenueMetric(SQLMetric): 

333 slug = "renewed_subscriptions_revenue" 

334 display_name = "Renewed Subscriptions Revenue" 

335 type = MetricType.currency 

336 query = MetricQuery.orders 

337 

338 @classmethod 

339 def get_sql_expression( 

340 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

341 ) -> ColumnElement[int]: 

342 return func.sum(Order.net_amount).filter( 

343 i.sql_date_trunc( 

344 cast(SQLColumnExpression[datetime], Subscription.started_at) 

345 ) 

346 != i.sql_date_trunc(t) 

347 ) 

348 

349 @classmethod 

350 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

351 return cumulative_sum(periods, cls.slug) 

352 

353 

354class RenewedSubscriptionsNetRevenueMetric(SQLMetric): 

355 slug = "renewed_subscriptions_net_revenue" 

356 display_name = "Renewed Subscriptions Net Revenue" 

357 type = MetricType.currency 

358 query = MetricQuery.orders 

359 

360 @classmethod 

361 def get_sql_expression( 

362 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

363 ) -> ColumnElement[int]: 

364 return func.sum(Order.payout_amount).filter( 

365 i.sql_date_trunc( 

366 cast(SQLColumnExpression[datetime], Subscription.started_at) 

367 ) 

368 != i.sql_date_trunc(t) 

369 ) 

370 

371 @classmethod 

372 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

373 return cumulative_sum(periods, cls.slug) 

374 

375 

376class ActiveSubscriptionsMetric(SQLMetric): 

377 slug = "active_subscriptions" 

378 display_name = "Active Subscriptions" 

379 type = MetricType.scalar 

380 query = MetricQuery.active_subscriptions 

381 

382 @classmethod 

383 def get_sql_expression( 

384 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

385 ) -> ColumnElement[int]: 

386 return func.count(Subscription.id) 

387 

388 @classmethod 

389 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

390 return cumulative_last(periods, cls.slug) 

391 

392 

393class MonthlyRecurringRevenueMetric(SQLMetric): 

394 slug = "monthly_recurring_revenue" 

395 display_name = "Monthly Recurring Revenue" 

396 type = MetricType.currency 

397 query = MetricQuery.active_subscriptions 

398 

399 @classmethod 

400 def get_sql_expression( 

401 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

402 ) -> ColumnElement[int]: 

403 return func.coalesce( 

404 func.sum( 

405 case( 

406 ( 

407 Subscription.recurring_interval 

408 == SubscriptionRecurringInterval.year, 

409 func.round(Subscription.amount / 12), 

410 ), 

411 ( 

412 Subscription.recurring_interval 

413 == SubscriptionRecurringInterval.month, 

414 Subscription.amount, 

415 ), 

416 ( 

417 Subscription.recurring_interval 

418 == SubscriptionRecurringInterval.week, 

419 func.round(Subscription.amount * 4), 

420 ), 

421 ( 

422 Subscription.recurring_interval 

423 == SubscriptionRecurringInterval.day, 

424 func.round(Subscription.amount * 30), 

425 ), 

426 ) 

427 ), 

428 0, 

429 ) 

430 

431 @classmethod 

432 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

433 return cumulative_last(periods, cls.slug) 

434 

435 

436class CommittedMonthlyRecurringRevenueMetric(SQLMetric): 

437 slug = "committed_monthly_recurring_revenue" 

438 display_name = "Committed Monthly Recurring Revenue" 

439 type = MetricType.currency 

440 query = MetricQuery.active_subscriptions 

441 

442 @classmethod 

443 def get_sql_expression( 

444 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

445 ) -> ColumnElement[int]: 

446 return func.coalesce( 

447 func.sum( 

448 case( 

449 ( 

450 Subscription.recurring_interval 

451 == SubscriptionRecurringInterval.year, 

452 func.round(Subscription.amount / 12), 

453 ), 

454 ( 

455 Subscription.recurring_interval 

456 == SubscriptionRecurringInterval.month, 

457 Subscription.amount, 

458 ), 

459 ( 

460 Subscription.recurring_interval 

461 == SubscriptionRecurringInterval.week, 

462 func.round(Subscription.amount * 4), 

463 ), 

464 ( 

465 Subscription.recurring_interval 

466 == SubscriptionRecurringInterval.day, 

467 func.round(Subscription.amount * 30), 

468 ), 

469 ) 

470 ).filter( 

471 or_( 

472 func.coalesce(Subscription.ended_at, Subscription.ends_at).is_( 

473 None 

474 ), 

475 i.sql_date_trunc( 

476 cast( 

477 SQLColumnExpression[datetime], 

478 func.coalesce(Subscription.ended_at, Subscription.ends_at), 

479 ) 

480 ) 

481 < i.sql_date_trunc(now), 

482 ) 

483 ), 

484 0, 

485 ) 

486 

487 @classmethod 

488 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

489 return cumulative_last(periods, cls.slug) 

490 

491 

492class CheckoutsMetric(SQLMetric): 

493 slug = "checkouts" 

494 display_name = "Checkouts" 

495 type = MetricType.scalar 

496 query = MetricQuery.checkouts 

497 

498 @classmethod 

499 def get_sql_expression( 

500 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

501 ) -> ColumnElement[int]: 

502 return func.count(Checkout.id) 

503 

504 @classmethod 

505 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

506 return cumulative_sum(periods, cls.slug) 

507 

508 

509class SucceededCheckoutsMetric(SQLMetric): 

510 slug = "succeeded_checkouts" 

511 display_name = "Succeeded Checkouts" 

512 type = MetricType.scalar 

513 query = MetricQuery.checkouts 

514 

515 @classmethod 

516 def get_sql_expression( 

517 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

518 ) -> ColumnElement[int]: 

519 return func.count(Checkout.id).filter( 

520 Checkout.status == CheckoutStatus.succeeded 

521 ) 

522 

523 @classmethod 

524 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

525 return cumulative_sum(periods, cls.slug) 

526 

527 

528class CheckoutsConversionMetric(SQLMetric): 

529 slug = "checkouts_conversion" 

530 display_name = "Checkouts Conversion Rate" 

531 type = MetricType.percentage 

532 query = MetricQuery.checkouts 

533 

534 @classmethod 

535 def get_sql_expression( 

536 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

537 ) -> ColumnElement[float]: 

538 return type_coerce( 

539 case( 

540 (func.count(Checkout.id) == 0, 0), 

541 else_=func.count(Checkout.id).filter( 

542 Checkout.status == CheckoutStatus.succeeded 

543 ) 

544 / func.count(Checkout.id), 

545 ), 

546 Float, 

547 ) 

548 

549 @classmethod 

550 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> float: 

551 total_checkouts = sum(getattr(p, "checkouts") for p in periods) 

552 total_succeeded = sum(getattr(p, "succeeded_checkouts") for p in periods) 

553 return total_succeeded / total_checkouts if total_checkouts > 0 else 0.0 

554 

555 

556class CanceledSubscriptionsMetric(SQLMetric): 

557 slug = "canceled_subscriptions" 

558 display_name = "Canceled Subscriptions" 

559 type = MetricType.scalar 

560 query = MetricQuery.canceled_subscriptions 

561 

562 @classmethod 

563 def get_sql_expression( 

564 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

565 ) -> ColumnElement[int]: 

566 return func.count(Subscription.id) 

567 

568 @classmethod 

569 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

570 return cumulative_sum(periods, cls.slug) 

571 

572 

573class CanceledSubscriptionsCustomerServiceMetric(SQLMetric): 

574 slug = "canceled_subscriptions_customer_service" 

575 display_name = "Canceled Subscriptions - Customer Service" 

576 type = MetricType.scalar 

577 query = MetricQuery.canceled_subscriptions 

578 

579 @classmethod 

580 def get_sql_expression( 

581 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

582 ) -> ColumnElement[int]: 

583 return func.count(Subscription.id).filter( 

584 Subscription.customer_cancellation_reason 

585 == CustomerCancellationReason.customer_service 

586 ) 

587 

588 @classmethod 

589 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

590 return cumulative_sum(periods, cls.slug) 

591 

592 

593class CanceledSubscriptionsLowQualityMetric(SQLMetric): 

594 slug = "canceled_subscriptions_low_quality" 

595 display_name = "Canceled Subscriptions - Low Quality" 

596 type = MetricType.scalar 

597 query = MetricQuery.canceled_subscriptions 

598 

599 @classmethod 

600 def get_sql_expression( 

601 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

602 ) -> ColumnElement[int]: 

603 return func.count(Subscription.id).filter( 

604 Subscription.customer_cancellation_reason 

605 == CustomerCancellationReason.low_quality 

606 ) 

607 

608 @classmethod 

609 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

610 return cumulative_sum(periods, cls.slug) 

611 

612 

613class CanceledSubscriptionsMissingFeaturesMetric(SQLMetric): 

614 slug = "canceled_subscriptions_missing_features" 

615 display_name = "Canceled Subscriptions - Missing Features" 

616 type = MetricType.scalar 

617 query = MetricQuery.canceled_subscriptions 

618 

619 @classmethod 

620 def get_sql_expression( 

621 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

622 ) -> ColumnElement[int]: 

623 return func.count(Subscription.id).filter( 

624 Subscription.customer_cancellation_reason 

625 == CustomerCancellationReason.missing_features 

626 ) 

627 

628 @classmethod 

629 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

630 return cumulative_sum(periods, cls.slug) 

631 

632 

633class CanceledSubscriptionsSwitchedServiceMetric(SQLMetric): 

634 slug = "canceled_subscriptions_switched_service" 

635 display_name = "Canceled Subscriptions - Switched Service" 

636 type = MetricType.scalar 

637 query = MetricQuery.canceled_subscriptions 

638 

639 @classmethod 

640 def get_sql_expression( 

641 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

642 ) -> ColumnElement[int]: 

643 return func.count(Subscription.id).filter( 

644 Subscription.customer_cancellation_reason 

645 == CustomerCancellationReason.switched_service 

646 ) 

647 

648 @classmethod 

649 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

650 return cumulative_sum(periods, cls.slug) 

651 

652 

653class CanceledSubscriptionsTooComplexMetric(SQLMetric): 

654 slug = "canceled_subscriptions_too_complex" 

655 display_name = "Canceled Subscriptions - Too Complex" 

656 type = MetricType.scalar 

657 query = MetricQuery.canceled_subscriptions 

658 

659 @classmethod 

660 def get_sql_expression( 

661 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

662 ) -> ColumnElement[int]: 

663 return func.count(Subscription.id).filter( 

664 Subscription.customer_cancellation_reason 

665 == CustomerCancellationReason.too_complex 

666 ) 

667 

668 @classmethod 

669 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

670 return cumulative_sum(periods, cls.slug) 

671 

672 

673class CanceledSubscriptionsTooExpensiveMetric(SQLMetric): 

674 slug = "canceled_subscriptions_too_expensive" 

675 display_name = "Canceled Subscriptions - Too Expensive" 

676 type = MetricType.scalar 

677 query = MetricQuery.canceled_subscriptions 

678 

679 @classmethod 

680 def get_sql_expression( 

681 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

682 ) -> ColumnElement[int]: 

683 return func.count(Subscription.id).filter( 

684 Subscription.customer_cancellation_reason 

685 == CustomerCancellationReason.too_expensive 

686 ) 

687 

688 @classmethod 

689 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

690 return cumulative_sum(periods, cls.slug) 

691 

692 

693class CanceledSubscriptionsUnusedMetric(SQLMetric): 

694 slug = "canceled_subscriptions_unused" 

695 display_name = "Canceled Subscriptions - Unused" 

696 type = MetricType.scalar 

697 query = MetricQuery.canceled_subscriptions 

698 

699 @classmethod 

700 def get_sql_expression( 

701 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

702 ) -> ColumnElement[int]: 

703 return func.count(Subscription.id).filter( 

704 Subscription.customer_cancellation_reason 

705 == CustomerCancellationReason.unused 

706 ) 

707 

708 @classmethod 

709 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

710 return cumulative_sum(periods, cls.slug) 

711 

712 

713class CanceledSubscriptionsOtherMetric(SQLMetric): 

714 slug = "canceled_subscriptions_other" 

715 display_name = "Canceled Subscriptions - Other" 

716 type = MetricType.scalar 

717 query = MetricQuery.canceled_subscriptions 

718 

719 @classmethod 

720 def get_sql_expression( 

721 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

722 ) -> ColumnElement[int]: 

723 return func.count(Subscription.id).filter( 

724 or_( 

725 Subscription.customer_cancellation_reason 

726 == CustomerCancellationReason.other, 

727 Subscription.customer_cancellation_reason.is_(None), 

728 ) 

729 ) 

730 

731 @classmethod 

732 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

733 return cumulative_sum(periods, cls.slug) 

734 

735 

736class ChurnedSubscriptionsMetric(SQLMetric): 

737 slug = "churned_subscriptions" 

738 display_name = "Churned Subscriptions" 

739 type = MetricType.scalar 

740 query = MetricQuery.churned_subscriptions 

741 

742 @classmethod 

743 def get_sql_expression( 

744 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

745 ) -> ColumnElement[int]: 

746 return func.count(Subscription.id) 

747 

748 @classmethod 

749 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

750 return cumulative_sum(periods, cls.slug) 

751 

752 

753class CostsMetric(SQLMetric): 

754 slug = "costs" 

755 display_name = "Costs" 

756 type = MetricType.currency_sub_cent 

757 query = MetricQuery.events 

758 

759 @classmethod 

760 def get_sql_expression( 

761 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

762 ) -> ColumnElement[int]: 

763 return func.sum( 

764 Event.user_metadata["_cost"]["amount"].as_numeric(17, 12) 

765 ).filter(Event.user_metadata["_cost"].is_not(None)) 

766 

767 @classmethod 

768 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

769 return cumulative_sum(periods, cls.slug) 

770 

771 

772class CumulativeCostsMetric(SQLMetric): 

773 slug = "cumulative_costs" 

774 display_name = "Cumulative Costs" 

775 type = MetricType.currency_sub_cent 

776 query = MetricQuery.events 

777 

778 @classmethod 

779 def get_sql_expression( 

780 cls, t: ColumnElement[datetime], i: TimeInterval, now: datetime 

781 ) -> ColumnElement[int]: 

782 return func.sum( 

783 Event.user_metadata["_cost"]["amount"].as_numeric(17, 12) 

784 ).filter(Event.user_metadata["_cost"].is_not(None)) 

785 

786 @classmethod 

787 def get_cumulative(cls, periods: Iterable["MetricsPeriod"]) -> int | float: 

788 return cumulative_last(periods, cls.slug) 

789